Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
209 changes: 0 additions & 209 deletions tests/integrations/fastmcp/test_fastmcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -934,77 +934,6 @@ def code_help_prompt(language: str):
assert SPANDATA.MCP_PROMPT_RESULT_MESSAGE_CONTENT not in span["data"]


@pytest.mark.parametrize("FastMCP", fastmcp_implementations, ids=fastmcp_ids)
@pytest.mark.asyncio
async def test_fastmcp_prompt_async(
sentry_init,
capture_events,
FastMCP,
json_rpc,
select_transactions_with_mcp_spans,
):
"""Test that FastMCP async prompt handlers create proper spans"""
sentry_init(
integrations=[MCPIntegration()],
traces_sample_rate=1.0,
)
events = capture_events()

mcp = FastMCP("Test Server")

session_manager = StreamableHTTPSessionManager(
app=mcp._mcp_server,
json_response=True,
)

app = Starlette(
routes=[
Mount("/mcp", app=session_manager.handle_request),
],
lifespan=lambda app: session_manager.run(),
)

# Try to register an async prompt handler
if hasattr(mcp, "prompt"):

@mcp.prompt()
async def async_prompt(topic: str):
"""Get async prompt for a topic"""
message1 = {
"role": "user",
"content": {"type": "text", "text": f"What is {topic}?"},
}

message2 = {
"role": "assistant",
"content": {
"type": "text",
"text": "Let me explain that",
},
}

if FASTMCP_VERSION is not None and FASTMCP_VERSION >= (3,):
message1 = Message(message1)
message2 = Message(message2)

return [message1, message2]

json_rpc(
app,
method="prompts/get",
params={
"name": "async_prompt",
"arguments": {"topic": "MCP"},
},
request_id="req-async-prompt",
)

transactions = select_transactions_with_mcp_spans(
events, method_name="prompts/get"
)
assert len(transactions) == 1


# =============================================================================
# Resource Handler Tests (if supported)
# =============================================================================
Expand Down Expand Up @@ -1599,144 +1528,6 @@ def stdio_tool(n: int) -> dict:
assert span["data"].get(SPANDATA.MCP_TRANSPORT) == "stdio"


# =============================================================================
# Integration-specific Tests
# =============================================================================


@pytest.mark.skipif(not HAS_MCP_FASTMCP, reason="mcp.server.fastmcp not installed")
def test_mcp_fastmcp_specific_features(sentry_init, capture_events):
"""Test features specific to mcp.server.fastmcp (from mcp package)"""
sentry_init(
integrations=[MCPIntegration()],
traces_sample_rate=1.0,
)
events = capture_events()

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("MCP Package Server")

@mcp.tool()
def package_specific_tool(x: int) -> int:
"""Tool for mcp.server.fastmcp package"""
return x + 100

with start_transaction(name="mcp.server.fastmcp tx"):
result = call_tool_through_mcp(mcp, "package_specific_tool", {"x": 50})

assert result["result"] == 150

(tx,) = events
assert tx["type"] == "transaction"


@pytest.mark.asyncio
@pytest.mark.skipif(
not HAS_STANDALONE_FASTMCP, reason="standalone fastmcp not installed"
)
async def test_standalone_fastmcp_specific_features(sentry_init, capture_events, stdio):
"""Test features specific to standalone fastmcp package"""
sentry_init(
integrations=[MCPIntegration()],
traces_sample_rate=1.0,
)
events = capture_events()

from fastmcp import FastMCP

mcp = FastMCP("Standalone FastMCP Server")

@mcp.tool()
def standalone_specific_tool(message: str) -> dict:
"""Tool for standalone fastmcp package"""
return {"echo": message, "length": len(message)}

with start_transaction(name="standalone fastmcp tx"):
await stdio(
mcp._mcp_server,
method="tools/call",
params={
"name": "standalone_specific_tool",
"arguments": {"message": "Hello FastMCP"},
},
)

(tx,) = events
assert tx["type"] == "transaction"


# =============================================================================
# Edge Cases and Robustness Tests
# =============================================================================


@pytest.mark.asyncio
@pytest.mark.parametrize("FastMCP", fastmcp_implementations, ids=fastmcp_ids)
async def test_fastmcp_tool_with_no_arguments(
sentry_init, capture_events, FastMCP, stdio
):
"""Test FastMCP tool with no arguments"""
sentry_init(
integrations=[MCPIntegration()],
traces_sample_rate=1.0,
)
events = capture_events()

mcp = FastMCP("Test Server")

@mcp.tool()
def no_args_tool() -> str:
"""Tool that takes no arguments"""
return "success"

with start_transaction(name="fastmcp tx"):
await stdio(
mcp._mcp_server,
method="tools/call",
params={
"name": "no_args_tool",
"arguments": {},
},
)

(tx,) = events
assert tx["type"] == "transaction"


@pytest.mark.asyncio
@pytest.mark.parametrize("FastMCP", fastmcp_implementations, ids=fastmcp_ids)
async def test_fastmcp_tool_with_none_return(
sentry_init, capture_events, FastMCP, stdio
):
"""Test FastMCP tool that returns None"""
sentry_init(
integrations=[MCPIntegration()],
traces_sample_rate=1.0,
)
events = capture_events()

mcp = FastMCP("Test Server")

@mcp.tool()
def none_return_tool(action: str) -> None:
"""Tool that returns None"""
pass

with start_transaction(name="fastmcp tx"):
await stdio(
mcp._mcp_server,
method="tools/call",
params={
"name": "none_return_tool",
"arguments": {"action": "log"},
},
)

(tx,) = events
assert tx["type"] == "transaction"


@pytest.mark.asyncio
@pytest.mark.parametrize("FastMCP", fastmcp_implementations, ids=fastmcp_ids)
@pytest.mark.parametrize("span_streaming", [True, False])
Expand Down
Loading