Skip to content

test(fastmcp): Replace broad try...except with skipif #6922

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alexander-alderman-webb merged 4 commits into from
Jul 29, 2026
+153 −157
Merged

test(fastmcp): Replace broad try...except with skipif #6922

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8028564
test(fastmcp): Replace broad try...except with skipif
alexander-alderman-webb Jul 29, 2026
4dafa5f
.
alexander-alderman-webb Jul 29, 2026
eb76bfa
.
alexander-alderman-webb Jul 29, 2026
dcf2d7a
fix boolean logic
alexander-alderman-webb Jul 29, 2026
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
310 changes: 153 additions & 157 deletions tests/integrations/fastmcp/test_fastmcp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ async def __call__(self, *args, **kwargs):
from sentry_sdk import start_transaction
from sentry_sdk.consts import OP, SPANDATA
from sentry_sdk.integrations.mcp import MCPIntegration
from sentry_sdk.utils import parse_version

try:
from fastmcp.prompts import Message
Expand Down Expand Up @@ -89,7 +90,9 @@ async def __call__(self, *args, **kwargs):
ReadResourceRequest = None

try:
from fastmcp import __version__ as FASTMCP_VERSION
from fastmcp import __version__

FASTMCP_VERSION = parse_version(__version__)
except ImportError:
FASTMCP_VERSION = None

Expand Down Expand Up @@ -885,7 +888,7 @@ def code_help_prompt(language: str):
},
}

if FASTMCP_VERSION is not None and FASTMCP_VERSION.startswith("3"):
if FASTMCP_VERSION is not None and FASTMCP_VERSION >= (3,):
message = Message(message)

return [message]
Expand Down Expand Up @@ -1014,7 +1017,7 @@ async def async_prompt(topic: str):
},
}

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

Expand Down Expand Up @@ -1043,6 +1046,10 @@ async def async_prompt(topic: str):
# =============================================================================


@pytest.mark.skipif(
HAS_STANDALONE_FASTMCP and (FASTMCP_VERSION is None or FASTMCP_VERSION < (0, 4, 1)),
reason="Resource URI templates not supported before fastmcp 0.4.1",
)
Comment thread
alexander-alderman-webb marked this conversation as resolved.
Comment thread
alexander-alderman-webb marked this conversation as resolved.
@pytest.mark.asyncio
@pytest.mark.parametrize("FastMCP", fastmcp_implementations, ids=fastmcp_ids)
@pytest.mark.parametrize("span_streaming", [True, False])
Expand All @@ -1064,93 +1071,86 @@ async def test_fastmcp_resource_sync(
mcp = FastMCP("Test Server")

# Try to register a resource handler
try:
if hasattr(mcp, "resource"):

@mcp.resource("file:///{path}")
def read_file(path: str):
"""Read a file resource"""
return "file contents"

if span_streaming:
items = capture_items("span")
with sentry_sdk.traces.start_span(name="custom parent"):
try:
result = await stdio(
mcp._mcp_server,
method="resources/read",
params={
"uri": "file:///test.txt",
},
request_id="req-resource",
if hasattr(mcp, "resource"):

@mcp.resource("file:///{path}")
def read_file(path: str):
"""Read a file resource"""
return "file contents"

if span_streaming:
items = capture_items("span")
with sentry_sdk.traces.start_span(name="custom parent"):
try:
result = await stdio(
mcp._mcp_server,
method="resources/read",
params={
"uri": "file:///test.txt",
},
request_id="req-resource",
)
except ValueError as e:
# Older FastMCP versions may not support this URI pattern
if "Unknown resource" in str(e):
pytest.skip(
f"Resource URI not supported in this FastMCP version: {e}"
)
except ValueError as e:
# Older FastMCP versions may not support this URI pattern
if "Unknown resource" in str(e):
pytest.skip(
f"Resource URI not supported in this FastMCP version: {e}"
)
raise

# Resource content is returned as-is
assert (
"file contents" in result.message.root.result["contents"][0]["text"]
)
raise

sentry_sdk.flush()
# Verify resource span was created
spans = [item.payload for item in items]
resource_spans = [
s
for s in spans
if s["attributes"].get("sentry.op") == OP.MCP_SERVER
]
assert len(resource_spans) == 1
span = resource_spans[0]
assert span["attributes"]["sentry.origin"] == "auto.ai.mcp"
assert span["name"] == "resources/read file:///test.txt"
assert span["attributes"][SPANDATA.MCP_RESOURCE_PROTOCOL] == "file"
else:
events = capture_events()
with start_transaction(name="fastmcp tx"):
try:
result = await stdio(
mcp._mcp_server,
method="resources/read",
params={
"uri": "file:///test.txt",
},
request_id="req-resource",
# Resource content is returned as-is
assert "file contents" in result.message.root.result["contents"][0]["text"]

sentry_sdk.flush()
# Verify resource span was created
spans = [item.payload for item in items]
resource_spans = [
s for s in spans if s["attributes"].get("sentry.op") == OP.MCP_SERVER
]
assert len(resource_spans) == 1
span = resource_spans[0]
assert span["attributes"]["sentry.origin"] == "auto.ai.mcp"
assert span["name"] == "resources/read file:///test.txt"
assert span["attributes"][SPANDATA.MCP_RESOURCE_PROTOCOL] == "file"
else:
events = capture_events()
with start_transaction(name="fastmcp tx"):
try:
result = await stdio(
mcp._mcp_server,
method="resources/read",
params={
"uri": "file:///test.txt",
},
request_id="req-resource",
)
except ValueError as e:
# Older FastMCP versions may not support this URI pattern
if "Unknown resource" in str(e):
pytest.skip(
f"Resource URI not supported in this FastMCP version: {e}"
)
except ValueError as e:
# Older FastMCP versions may not support this URI pattern
if "Unknown resource" in str(e):
pytest.skip(
f"Resource URI not supported in this FastMCP version: {e}"
)
raise

# Resource content is returned as-is
assert (
"file contents" in result.message.root.result["contents"][0]["text"]
)
raise

(tx,) = events
assert tx["type"] == "transaction"
# Resource content is returned as-is
assert "file contents" in result.message.root.result["contents"][0]["text"]

# Verify resource span was created
resource_spans = [s for s in tx["spans"] if s["op"] == OP.MCP_SERVER]
assert len(resource_spans) == 1
span = resource_spans[0]
assert span["origin"] == "auto.ai.mcp"
assert span["description"] == "resources/read file:///test.txt"
assert span["data"][SPANDATA.MCP_RESOURCE_PROTOCOL] == "file"
(tx,) = events
assert tx["type"] == "transaction"

except (AttributeError, TypeError):
# Resource handler not supported in this version
pytest.skip("Resource handlers not supported in this FastMCP version")
# Verify resource span was created
resource_spans = [s for s in tx["spans"] if s["op"] == OP.MCP_SERVER]
assert len(resource_spans) == 1
span = resource_spans[0]
assert span["origin"] == "auto.ai.mcp"
assert span["description"] == "resources/read file:///test.txt"
assert span["data"][SPANDATA.MCP_RESOURCE_PROTOCOL] == "file"


@pytest.mark.skipif(
HAS_STANDALONE_FASTMCP and (FASTMCP_VERSION is None or FASTMCP_VERSION < (0, 4, 1)),
reason="Resource URI templates not supported before fastmcp 0.4.1",
)
@pytest.mark.parametrize("FastMCP", fastmcp_implementations, ids=fastmcp_ids)
@pytest.mark.asyncio
@pytest.mark.parametrize("span_streaming", [True, False])
Expand Down Expand Up @@ -1185,83 +1185,79 @@ async def test_fastmcp_resource_async(
)

# Try to register an async resource handler
try:
if hasattr(mcp, "resource"):
if span_streaming:
items = capture_items("span")

@mcp.resource("https://example.com/{resource}")
async def read_url(resource: str):
"""Read a URL resource"""
return "resource data"

_, result = json_rpc(
app,
method="resources/read",
params={
"uri": "https://example.com/resource",
},
request_id="req-async-resource",
)
# Older FastMCP versions may not support this URI pattern
if (
"error" in result.json()
and "Unknown resource" in result.json()["error"]["message"]
):
pytest.skip("Resource URI not supported in this FastMCP version.")
return

assert "resource data" in result.json()["result"]["contents"][0]["text"]

sentry_sdk.flush()
spans = [item.payload for item in items]
spans = [
span
for span in spans
if span["attributes"].get("mcp.method.name") == "resources/read"
]
assert len(spans) == 1
span = spans[0]

assert span["attributes"][SPANDATA.MCP_RESOURCE_PROTOCOL] == "https"
else:
events = capture_events()
if hasattr(mcp, "resource"):
if span_streaming:
items = capture_items("span")

@mcp.resource("https://example.com/{resource}")
async def read_url(resource: str):
"""Read a URL resource"""
return "resource data"
@mcp.resource("https://example.com/{resource}")
async def read_url(resource: str):
"""Read a URL resource"""
return "resource data"

_, result = json_rpc(
app,
method="resources/read",
params={
"uri": "https://example.com/resource",
},
request_id="req-async-resource",
)
# Older FastMCP versions may not support this URI pattern
if (
"error" in result.json()
and "Unknown resource" in result.json()["error"]["message"]
):
pytest.skip("Resource URI not supported in this FastMCP version.")
return

assert "resource data" in result.json()["result"]["contents"][0]["text"]

transactions = select_transactions_with_mcp_spans(
events, method_name="resources/read"
)
assert len(transactions) == 1
tx = transactions[0]
assert len(tx["spans"]) == 1
span = tx["spans"][0]

assert span["data"][SPANDATA.MCP_RESOURCE_PROTOCOL] == "https"
except (AttributeError, TypeError):
# Resource handler not supported in this version
pytest.skip("Resource handlers not supported in this FastMCP version")
_, result = json_rpc(
app,
method="resources/read",
params={
"uri": "https://example.com/resource",
},
request_id="req-async-resource",
)
# Older FastMCP versions may not support this URI pattern
if (
"error" in result.json()
and "Unknown resource" in result.json()["error"]["message"]
):
pytest.skip("Resource URI not supported in this FastMCP version.")
return

assert "resource data" in result.json()["result"]["contents"][0]["text"]

sentry_sdk.flush()
spans = [item.payload for item in items]
spans = [
span
for span in spans
if span["attributes"].get("mcp.method.name") == "resources/read"
]
assert len(spans) == 1
span = spans[0]

assert span["attributes"][SPANDATA.MCP_RESOURCE_PROTOCOL] == "https"
else:
events = capture_events()

@mcp.resource("https://example.com/{resource}")
async def read_url(resource: str):
"""Read a URL resource"""
return "resource data"

_, result = json_rpc(
app,
method="resources/read",
params={
"uri": "https://example.com/resource",
},
request_id="req-async-resource",
)
# Older FastMCP versions may not support this URI pattern
if (
"error" in result.json()
and "Unknown resource" in result.json()["error"]["message"]
):
pytest.skip("Resource URI not supported in this FastMCP version.")
return

assert "resource data" in result.json()["result"]["contents"][0]["text"]

transactions = select_transactions_with_mcp_spans(
events, method_name="resources/read"
)
assert len(transactions) == 1
tx = transactions[0]
assert len(tx["spans"]) == 1
span = tx["spans"][0]

assert span["data"][SPANDATA.MCP_RESOURCE_PROTOCOL] == "https"


# =============================================================================
Expand Down
Loading