diff --git a/tests/integrations/fastmcp/test_fastmcp.py b/tests/integrations/fastmcp/test_fastmcp.py index 71d780fa04..6b2dc3c643 100644 --- a/tests/integrations/fastmcp/test_fastmcp.py +++ b/tests/integrations/fastmcp/test_fastmcp.py @@ -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) # ============================================================================= @@ -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])