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

Fix event loop conflict in inspect CLI command

Browse source 
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Jeremiah Lowin 2025-06-21 20:04:39 -04:00
commit 8506f78453
4 changed files with 41 additions and 401 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -3,7 +3,6 @@
import asyncio
import importlib.metadata
import importlib.util
import json
import os
import platform
import subprocess
@ -13,6 +12,7 @@ from typing import Annotated
import dotenv
import typer
from pydantic import TypeAdapter
from rich.console import Console
from rich.table import Table
from typer import Context, Exit
@ -21,7 +21,7 @@ import fastmcp
from fastmcp.cli import claude
from fastmcp.cli import run as run_module
from fastmcp.server.server import FastMCP
from fastmcp.utilities.inspect import get_fastmcp_info
from fastmcp.utilities.inspect import FastMCPInfo, inspect_fastmcp
from fastmcp.utilities.logging import get_logger
logger = get_logger("cli")
@ -487,35 +487,29 @@ def inspect(
# Get server information
async def get_info():
return await get_fastmcp_info(server)
return await inspect_fastmcp(server)
info = asyncio.run(get_info())
try:
# Try to use existing event loop if available
asyncio.get_running_loop()
# If there's already a loop running, we need to run in a thread
import concurrent.futures
# Convert to dict for JSON serialization
from dataclasses import asdict
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(asyncio.run, get_info())
info = future.result()
except RuntimeError:
# No running loop, safe to use asyncio.run
info = asyncio.run(get_info())
def convert_for_json(obj):
"""Convert objects for JSON serialization."""
if isinstance(obj, list):
return [convert_for_json(item) for item in obj]
elif isinstance(obj, set):
return list(obj)
elif hasattr(obj, "model_dump"): # Pydantic models
return obj.model_dump()
else:
return obj
info_dict = asdict(
info,
dict_factory=lambda fields: {k: convert_for_json(v) for k, v in fields},
)
info_json = TypeAdapter(FastMCPInfo).dump_json(info, indent=2)
# Ensure output directory exists
output.parent.mkdir(parents=True, exist_ok=True)
# Write JSON report (always pretty-printed)
with output.open("w", encoding="utf-8") as f:
json.dump(info_dict, f, indent=2, ensure_ascii=False)
f.write(info_json.decode("utf-8"))
logger.info(f"Server inspection complete. Report saved to {output}")

10
src/fastmcp/utilities/inspect.py
View file

@ -79,7 +79,7 @@ class FastMCPInfo:
capabilities: dict[str, Any]
async def get_fastmcp_info_v2(mcp: FastMCP[Any]) -> FastMCPInfo:
async def inspect_fastmcp_v2(mcp: FastMCP[Any]) -> FastMCPInfo:
"""Extract information from a FastMCP v2.x instance.
Args:
@ -179,7 +179,7 @@ async def get_fastmcp_info_v2(mcp: FastMCP[Any]) -> FastMCPInfo:
)
async def get_fastmcp_info_v1(mcp: Any) -> FastMCPInfo:
async def inspect_fastmcp_v1(mcp: Any) -> FastMCPInfo:
"""Extract information from a FastMCP v1.x instance using a Client.
Args:
@ -308,7 +308,7 @@ def _is_fastmcp_v1(mcp: Any) -> bool:
return isinstance(mcp, FastMCP1x) and not isinstance(mcp, FastMCP)
async def get_fastmcp_info(mcp: FastMCP[Any] | Any) -> FastMCPInfo:
async def inspect_fastmcp(mcp: FastMCP[Any] | Any) -> FastMCPInfo:
"""Extract information from a FastMCP instance into a dataclass.
This function automatically detects whether the instance is FastMCP v1.x or v2.x
@ -321,6 +321,6 @@ async def get_fastmcp_info(mcp: FastMCP[Any] | Any) -> FastMCPInfo:
FastMCPInfo dataclass containing the extracted information
"""
if _is_fastmcp_v1(mcp):
return await get_fastmcp_info_v1(mcp)
return await inspect_fastmcp_v1(mcp)
else:
return await get_fastmcp_info_v2(mcp)
return await inspect_fastmcp_v2(mcp)

354
tests/cli/test_inspect.py
View file

@ -1,354 +0,0 @@
"""Tests for the CLI inspect command."""
import json
import tempfile
from pathlib import Path
from typer.testing import CliRunner
from fastmcp.cli.cli import app
class TestInspectCommand:
"""Tests for the fastmcp inspect CLI command."""
def setup_method(self):
"""Set up test fixtures."""
self.runner = CliRunner()
def test_inspect_basic_server(self):
"""Test inspecting a basic FastMCP 2.x server."""
# Create a temporary server file
server_content = '''
from fastmcp import FastMCP
mcp = FastMCP("TestServer", instructions="A test server")
@mcp.tool
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
@mcp.resource("resource://data")
def get_data() -> str:
"""Get test data."""
return "test data"
@mcp.prompt
def test_prompt(message: str) -> list:
"""Test prompt."""
return [{"role": "user", "content": message}]
'''
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(server_content)
server_file = f.name
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
output_file = f.name
try:
# Run the inspect command
result = self.runner.invoke(
app, ["inspect", server_file, "-o", output_file]
)
assert result.exit_code == 0
assert "✓ Inspected server: TestServer" in result.stdout
assert "Tools: 1" in result.stdout
assert "Prompts: 1" in result.stdout
assert "Resources: 1" in result.stdout
# Check the JSON output
with open(output_file) as f:
data = json.load(f)
assert data["name"] == "TestServer"
assert data["instructions"] == "A test server"
assert "fastmcp_version" in data
assert "mcp_version" in data
assert "server_version" in data
# Check tools
assert len(data["tools"]) == 1
tool = data["tools"][0]
assert tool["key"] == "add"
assert tool["name"] == "add"
assert tool["description"] == "Add two numbers."
assert "input_schema" in tool
assert tool["enabled"] is True
# Check resources
assert len(data["resources"]) == 1
resource = data["resources"][0]
assert resource["key"] == "resource://data"
assert resource["uri"] == "resource://data"
assert resource["name"] == "get_data"
# Check prompts
assert len(data["prompts"]) == 1
prompt = data["prompts"][0]
assert prompt["key"] == "test_prompt"
assert prompt["name"] == "test_prompt"
assert prompt["description"] == "Test prompt."
# Check capabilities
assert "capabilities" in data
assert "tools" in data["capabilities"]
finally:
# Clean up
Path(server_file).unlink(missing_ok=True)
Path(output_file).unlink(missing_ok=True)
def test_inspect_with_object_spec(self):
"""Test inspecting a server with object specification."""
server_content = '''
from fastmcp import FastMCP
server = FastMCP("ObjectSpecServer")
@server.tool
def multiply(a: int, b: int) -> int:
"""Multiply two numbers."""
return a * b
'''
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(server_content)
server_file = f.name
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
output_file = f.name
try:
# Run the inspect command with object specification
result = self.runner.invoke(
app, ["inspect", f"{server_file}:server", "-o", output_file]
)
assert result.exit_code == 0
assert "✓ Inspected server: ObjectSpecServer" in result.stdout
# Check the JSON output
with open(output_file) as f:
data = json.load(f)
assert data["name"] == "ObjectSpecServer"
assert len(data["tools"]) == 1
assert data["tools"][0]["name"] == "multiply"
finally:
# Clean up
Path(server_file).unlink(missing_ok=True)
Path(output_file).unlink(missing_ok=True)
def test_inspect_default_output(self):
"""Test inspecting with default output filename."""
server_content = '''
from fastmcp import FastMCP
mcp = FastMCP("DefaultOutputServer")
@mcp.tool
def test_tool() -> str:
"""Test tool."""
return "test"
'''
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(server_content)
server_file = f.name
try:
# Run the inspect command without specifying output file
result = self.runner.invoke(app, ["inspect", server_file])
assert result.exit_code == 0
assert "✓ Inspected server: DefaultOutputServer" in result.stdout
assert "Report saved to: server-info.json" in result.stdout
# Check the default output file exists
default_output = Path("server-info.json")
assert default_output.exists()
# Check the JSON content
with open(default_output) as f:
data = json.load(f)
assert data["name"] == "DefaultOutputServer"
finally:
# Clean up
Path(server_file).unlink(missing_ok=True)
Path("server-info.json").unlink(missing_ok=True)
def test_inspect_invalid_server_file(self):
"""Test inspecting a non-existent server file."""
result = self.runner.invoke(
app, ["inspect", "nonexistent.py", "-o", "output.json"]
)
assert result.exit_code == 1
# The error happens at the file parsing level, so no stdout output
def test_inspect_server_with_error(self):
"""Test inspecting a server file with syntax errors."""
server_content = """
from fastmcp import FastMCP
mcp = FastMCP("ErrorServer")
# Syntax error below
@mcp.tool
def broken_tool(
"""
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(server_content)
server_file = f.name
try:
result = self.runner.invoke(
app, ["inspect", server_file, "-o", "output.json"]
)
assert result.exit_code == 1
assert "✗ Failed to inspect server:" in result.stdout
finally:
# Clean up
Path(server_file).unlink(missing_ok=True)
Path("output.json").unlink(missing_ok=True)
def test_inspect_comprehensive_json_structure(self):
"""Test that the JSON output has the correct structure."""
server_content = '''
from fastmcp import FastMCP
mcp = FastMCP("ComprehensiveServer", instructions="Full test server")
@mcp.tool
def calculate(x: int, y: int) -> int:
"""Calculate something."""
return x + y
@mcp.resource("resource://static")
def static_resource() -> str:
"""Static resource."""
return "static"
@mcp.resource("resource://template/{id}")
def template_resource(id: str) -> str:
"""Template resource."""
return f"data-{id}"
@mcp.prompt
def analysis_prompt(data: str) -> list:
"""Analysis prompt."""
return [{"role": "user", "content": f"Analyze: {data}"}]
'''
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(server_content)
server_file = f.name
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
output_file = f.name
try:
result = self.runner.invoke(
app, ["inspect", server_file, "-o", output_file]
)
assert result.exit_code == 0
# Load and validate JSON structure
with open(output_file) as f:
data = json.load(f)
# Check top-level structure
required_fields = [
"name",
"instructions",
"fastmcp_version",
"mcp_version",
"server_version",
"tools",
"prompts",
"resources",
"templates",
"capabilities",
]
for field in required_fields:
assert field in data, f"Missing field: {field}"
# Check version fields are strings
assert isinstance(data["fastmcp_version"], str)
assert isinstance(data["mcp_version"], str)
assert isinstance(data["server_version"], str)
# Check that we have the expected components
assert len(data["tools"]) == 1
assert len(data["resources"]) == 1
assert len(data["templates"]) == 1
assert len(data["prompts"]) == 1
# Check tool structure
tool = data["tools"][0]
tool_fields = [
"key",
"name",
"description",
"input_schema",
"annotations",
"tags",
"enabled",
]
for field in tool_fields:
assert field in tool, f"Missing tool field: {field}"
# Check resource structure
resource = data["resources"][0]
resource_fields = [
"key",
"uri",
"name",
"description",
"mime_type",
"tags",
"enabled",
]
for field in resource_fields:
assert field in resource, f"Missing resource field: {field}"
# Check template structure
template = data["templates"][0]
template_fields = [
"key",
"uri_template",
"name",
"description",
"mime_type",
"tags",
"enabled",
]
for field in template_fields:
assert field in template, f"Missing template field: {field}"
# Check prompt structure
prompt = data["prompts"][0]
prompt_fields = [
"key",
"name",
"description",
"arguments",
"tags",
"enabled",
]
for field in prompt_fields:
assert field in prompt, f"Missing prompt field: {field}"
finally:
# Clean up
Path(server_file).unlink(missing_ok=True)
Path(output_file).unlink(missing_ok=True)

34
tests/utilities/test_inspect.py
View file

@ -9,8 +9,8 @@ from fastmcp.utilities.inspect import (
FastMCPInfo,
ToolInfo,
_is_fastmcp_v1,
get_fastmcp_info,
get_fastmcp_info_v1,
inspect_fastmcp,
inspect_fastmcp_v1,
)
@ -69,7 +69,7 @@ class TestGetFastMCPInfo:
"""Test get_fastmcp_info with an empty server."""
mcp = FastMCP("EmptyServer", instructions="Empty server for testing")
info = await get_fastmcp_info(mcp)
info = await inspect_fastmcp(mcp)
assert info.name == "EmptyServer"
assert info.instructions == "Empty server for testing"
@ -97,7 +97,7 @@ class TestGetFastMCPInfo:
def greet(name: str) -> str:
return f"Hello, {name}!"
info = await get_fastmcp_info(mcp)
info = await inspect_fastmcp(mcp)
assert info.name == "ToolServer"
assert len(info.tools) == 2
@ -117,7 +117,7 @@ class TestGetFastMCPInfo:
def get_dynamic_data(param: str) -> str:
return f"Dynamic data: {param}"
info = await get_fastmcp_info(mcp)
info = await inspect_fastmcp(mcp)
assert info.name == "ResourceServer"
assert len(info.resources) == 1 # Static resource
@ -139,7 +139,7 @@ class TestGetFastMCPInfo:
def custom_analysis(text: str) -> list:
return [{"role": "user", "content": f"Custom: {text}"}]
info = await get_fastmcp_info(mcp)
info = await inspect_fastmcp(mcp)
assert info.name == "PromptServer"
assert len(info.prompts) == 2
@ -171,7 +171,7 @@ class TestGetFastMCPInfo:
def analyze(content: str) -> list:
return [{"role": "user", "content": content}]
info = await get_fastmcp_info(mcp)
info = await inspect_fastmcp(mcp)
assert info.name == "ComprehensiveServer"
assert info.instructions == "A server with everything"
@ -204,7 +204,7 @@ class TestGetFastMCPInfo:
"""Test get_fastmcp_info with a server that has no instructions."""
mcp = FastMCP("NoInstructionsServer")
info = await get_fastmcp_info(mcp)
info = await inspect_fastmcp(mcp)
assert info.name == "NoInstructionsServer"
assert info.instructions is None
@ -226,7 +226,7 @@ class TestGetFastMCPInfo:
return [{"role": "user", "content": "test"}]
# Get info using our function
info = await get_fastmcp_info(mcp)
info = await inspect_fastmcp(mcp)
# Verify using client
async with Client(mcp) as client:
@ -258,7 +258,7 @@ class TestFastMCP1xCompatibility:
"""Test get_fastmcp_info_v1 with an empty FastMCP1x server."""
mcp = FastMCP1x("Test1x")
info = await get_fastmcp_info_v1(mcp)
info = await inspect_fastmcp_v1(mcp)
assert info.name == "Test1x"
assert info.instructions is None
@ -283,7 +283,7 @@ class TestFastMCP1xCompatibility:
def greet(name: str) -> str:
return f"Hello, {name}!"
info = await get_fastmcp_info_v1(mcp)
info = await inspect_fastmcp_v1(mcp)
assert info.name == "Test1x"
assert len(info.tools) == 2
@ -299,7 +299,7 @@ class TestFastMCP1xCompatibility:
def get_data() -> str:
return "Some data"
info = await get_fastmcp_info_v1(mcp)
info = await inspect_fastmcp_v1(mcp)
assert info.name == "Test1x"
assert len(info.resources) == 1
@ -315,7 +315,7 @@ class TestFastMCP1xCompatibility:
def analyze_data(data: str) -> list:
return [{"role": "user", "content": f"Analyze: {data}"}]
info = await get_fastmcp_info_v1(mcp)
info = await inspect_fastmcp_v1(mcp)
assert info.name == "Test1x"
assert len(info.prompts) == 1
@ -330,7 +330,7 @@ class TestFastMCP1xCompatibility:
def test_tool() -> str:
return "test"
info = await get_fastmcp_info(mcp)
info = await inspect_fastmcp(mcp)
assert info.name == "Test1x"
assert len(info.tools) == 1
@ -346,7 +346,7 @@ class TestFastMCP1xCompatibility:
def test_tool() -> str:
return "test"
info = await get_fastmcp_info(mcp)
info = await inspect_fastmcp(mcp)
assert info.name == "Test2x"
assert len(info.tools) == 1
@ -366,8 +366,8 @@ class TestFastMCP1xCompatibility:
def tool2x() -> str:
return "2x"
info1x = await get_fastmcp_info(mcp1x)
info2x = await get_fastmcp_info(mcp2x)
info1x = await inspect_fastmcp(mcp1x)
info2x = await inspect_fastmcp(mcp2x)
assert info1x.name == "Test1x"
assert info2x.name == "Test2x"