Repository: jlowin/fastmcp
Pull request #255 (2025-04-25 16:55:40)

Add resource template tests


Patch:
diff --git a/src/fastmcp/server/openapi.py b/src/fastmcp/server/openapi.py
index 0911cae..bc65c2f 100644
--- a/src/fastmcp/server/openapi.py
+++ b/src/fastmcp/server/openapi.py
@@ -257,7 +257,7 @@ class OpenAPIResource(Resource):
         self._client = client
         self._route = route
 
-    async def read(self) -> str | bytes:
+    async def read(self) -> str:
         """Fetch the resource data by making an HTTP request."""
         try:
             # Extract path parameters from the URI if present
@@ -297,16 +297,15 @@ class OpenAPIResource(Resource):
             # Raise for 4xx/5xx responses
             response.raise_for_status()
 
-            # Determine content type and return appropriate format
-            content_type = response.headers.get("content-type", "").lower()
-
-            if "application/json" in content_type:
-                result = response.json()
-                return json.dumps(result)
-            elif any(ct in content_type for ct in ["text/", "application/xml"]):
-                return response.text
+            # Return response content based on mime type
+            if self.mime_type == "application/json":
+                try:
+                    return response.json()
+                except (json.JSONDecodeError, ValueError):
+                    # Fallback to returning the text
+                    return response.text
             else:
-                return response.content
+                return response.text
 
         except httpx.HTTPStatusError as e:
             # Handle HTTP errors (4xx, 5xx)
@@ -344,13 +343,59 @@ class OpenAPIResourceTemplate(ResourceTemplate):
             uri_template=uri_template,
             name=name,
             description=description,
-            fn=lambda **kwargs: None,
+            fn=self._create_resource_fn,
             parameters=parameters,
             tags=tags,
         )
         self._client = client
         self._route = route
 
+    async def _create_resource_fn(self, **kwargs):
+        """Create a resource with parameters."""
+        # Prepare the path with parameters
+        path = self._route.path
+        for param_name, param_value in kwargs.items():
+            path = path.replace(f"{{{param_name}}}", str(param_value))
+
+        try:
+            response = await self._client.request(
+                method=self._route.method,
+                url=path,
+                timeout=30.0,  # Default timeout
+            )
+
+            # Raise for 4xx/5xx responses
+            response.raise_for_status()
+
+            # Determine the mime type from the response
+            content_type = response.headers.get("content-type", "application/json")
+            mime_type = content_type.split(";")[0].strip()
+
+            # Return the appropriate data
+            if mime_type == "application/json":
+                try:
+                    return response.json()
+                except (json.JSONDecodeError, ValueError):
+                    return response.text
+            else:
+                return response.text
+
+        except httpx.HTTPStatusError as e:
+            error_message = (
+                f"HTTP error {e.response.status_code}: {e.response.reason_phrase}"
+            )
+            try:
+                error_data = e.response.json()
+                error_message += f" - {error_data}"
+            except (json.JSONDecodeError, ValueError):
+                if e.response.text:
+                    error_message += f" - {e.response.text}"
+
+            raise ValueError(error_message)
+
+        except httpx.RequestError as e:
+            raise ValueError(f"Request error: {str(e)}")
+
     async def create_resource(self, uri: str, params: dict[str, Any]) -> Resource:
         """Create a resource with the given parameters."""
         # Generate a URI for this resource instance
@@ -364,8 +409,9 @@ class OpenAPIResourceTemplate(ResourceTemplate):
             route=self._route,
             uri=uri,
             name=f"{self.name}-{'-'.join(uri_parts)}",
-            description=self.description or f"Resource for {self._route.path}",
-            mime_type="application/json",
+            description=self.description
+            or f"Resource for {self._route.path}",  # Provide default if None
+            mime_type="application/json",  # Default, will be updated when read
             tags=set(self._route.tags or []),
         )
 


Test patch:
diff --git a/tests/resources/test_resource_template.py b/tests/resources/test_resource_template.py
index 38b2f45..33db36c 100644
--- a/tests/resources/test_resource_template.py
+++ b/tests/resources/test_resource_template.py
@@ -443,3 +443,20 @@ class TestMatchUriTemplate:
         uri_template = "test://a/{x}{y}"
         result = match_uri_template(uri=uri, uri_template=uri_template)
         assert result is None
+
+    @pytest.mark.parametrize(
+        "uri, expected_params",
+        [
+            ("file://abc/xyz.py", {"path": "xyz"}),
+            ("file://abc/x/y/z.py", {"path": "x/y/z"}),
+            ("file://abc/x/y/z/.py", {"path": "x/y/z/"}),
+            ("file://abc/x/y/z.md", None),
+            ("file://x/y/z.txt", None),
+        ],
+    )
+    def test_match_uri_template_with_non_slash_suffix(
+        self, uri: str, expected_params: dict[str, str]
+    ):
+        uri_template = "file://abc/{path*}.py"
+        result = match_uri_template(uri=uri, uri_template=uri_template)
+        assert result == expected_params
diff --git a/tests/server/test_openapi.py b/tests/server/test_openapi.py
index 8b6e828..afad3b6 100644
--- a/tests/server/test_openapi.py
+++ b/tests/server/test_openapi.py
@@ -1,14 +1,12 @@
-import base64
 import json
 import re
 
 import httpx
 import pytest
 from dirty_equals import IsStr
-from fastapi import FastAPI, HTTPException, Response
-from fastapi.responses import PlainTextResponse
+from fastapi import FastAPI, HTTPException
 from httpx import ASGITransport, AsyncClient
-from mcp.types import BlobResourceContents, TextContent, TextResourceContents
+from mcp.types import TextContent
 from pydantic import BaseModel, TypeAdapter
 from pydantic.networks import AnyUrl
 
@@ -68,17 +66,6 @@ def fastapi_app(users_db: dict[int, User]) -> FastAPI:
         user.name = name
         return user
 
-    @app.get("/ping", response_class=PlainTextResponse)
-    async def ping() -> str:
-        """Ping the server."""
-        return "pong"
-
-    @app.get("/ping-bytes")
-    async def ping_bytes() -> Response:
-        """Ping the server and get a bytes response."""
-
-        return Response(content=b"pong")
-
     return app
 
 
@@ -133,8 +120,7 @@ class TestTools:
         """
         By default, tools exclude GET methods
         """
-        async with Client(fastmcp_openapi_server) as client:
-            tools = await client.list_tools()
+        tools = await fastmcp_openapi_server._mcp_list_tools()
         assert len(tools) == 2
 
         assert tools[0].model_dump() == dict(
@@ -170,10 +156,9 @@ class TestTools:
         """
         The tool created by the OpenAPI server should be the same as the original
         """
-        async with Client(fastmcp_openapi_server) as client:
-            tool_response = await client.call_tool(
-                "create_user_users_post", {"name": "David", "active": False}
-            )
+        tool_response = await fastmcp_openapi_server._mcp_call_tool(
+            "create_user_users_post", {"name": "David", "active": False}
+        )
 
         # Convert TextContent to dict for comparison
         assert isinstance(tool_response, list) and len(tool_response) == 1
@@ -188,13 +173,10 @@ class TestTools:
         assert len(response.json()) == 4
 
         # Check that the user was created via MCP
-        async with Client(fastmcp_openapi_server) as client:
-            user_response = await client.read_resource(
-                "resource://openapi/get_user_users__user_id__get/4"
-            )
-            assert isinstance(user_response[0], TextResourceContents)
-            response_text = user_response[0].text
-            user = json.loads(response_text)
+        user_response = await fastmcp_openapi_server._mcp_read_resource(
+            "resource://openapi/get_user_users__user_id__get/4"
+        )
+        user = user_response[0].content
         assert user == expected_user
 
     async def test_call_update_user_name_tool(
@@ -203,11 +185,9 @@ class TestTools:
         """
         The tool created by the OpenAPI server should be the same as the original
         """
-        async with Client(fastmcp_openapi_server) as client:
-            tool_response = await client.call_tool(
-                "update_user_name_users__user_id__name_patch",
-                {"user_id": 1, "name": "XYZ"},
-            )
+        tool_response = await fastmcp_openapi_server._mcp_call_tool(
+            "update_user_name_users__user_id__name_patch", {"user_id": 1, "name": "XYZ"}
+        )
 
         # Convert TextContent to dict for comparison
         assert isinstance(tool_response, list) and len(tool_response) == 1
@@ -222,13 +202,10 @@ class TestTools:
         assert expected_data in response.json()
 
         # Check that the user was updated via MCP
-        async with Client(fastmcp_openapi_server) as client:
-            user_response = await client.read_resource(
-                "resource://openapi/get_user_users__user_id__get/1"
-            )
-            assert isinstance(user_response[0], TextResourceContents)
-            response_text = user_response[0].text
-            user = json.loads(response_text)
+        user_response = await fastmcp_openapi_server._mcp_read_resource(
+            "resource://openapi/get_user_users__user_id__get/1"
+        )
+        user = user_response[0].content
         assert user == expected_data
 
 
@@ -237,9 +214,8 @@ class TestResources:
         """
         By default, resources exclude GET methods without parameters
         """
-        async with Client(fastmcp_openapi_server) as client:
-            resources = await client.list_resources()
-        assert len(resources) == 3
+        resources = await fastmcp_openapi_server._mcp_list_resources()
+        assert len(resources) == 1
         assert resources[0].uri == AnyUrl("resource://openapi/get_users_users_get")
         assert resources[0].name == "get_users_users_get"
 
@@ -252,47 +228,17 @@ class TestResources:
         """
         The resource created by the OpenAPI server should be the same as the original
         """
-
         json_users = TypeAdapter(list[User]).dump_python(
             sorted(users_db.values(), key=lambda x: x.id)
         )
-        async with Client(fastmcp_openapi_server) as client:
-            resource_response = await client.read_resource(
-                "resource://openapi/get_users_users_get"
-            )
-            assert isinstance(resource_response[0], TextResourceContents)
-            response_text = resource_response[0].text
-            resource = json.loads(response_text)
+        resource_response = await fastmcp_openapi_server._mcp_read_resource(
+            "resource://openapi/get_users_users_get"
+        )
+        resource = resource_response[0].content
         assert resource == json_users
         response = await api_client.get("/users")
         assert response.json() == json_users
 
-    async def test_get_bytes_resource(
-        self,
-        fastmcp_openapi_server: FastMCPOpenAPI,
-        api_client,
-    ):
-        """Test reading a resource that returns bytes."""
-        async with Client(fastmcp_openapi_server) as client:
-            resource_response = await client.read_resource(
-                "resource://openapi/ping_bytes_ping_bytes_get"
-            )
-            assert isinstance(resource_response[0], BlobResourceContents)
-            assert base64.b64decode(resource_response[0].blob) == b"pong"
-
-    async def test_get_str_resource(
-        self,
-        fastmcp_openapi_server: FastMCPOpenAPI,
-        api_client,
-    ):
-        """Test reading a resource that returns a string."""
-        async with Client(fastmcp_openapi_server) as client:
-            resource_response = await client.read_resource(
-                "resource://openapi/ping_ping_get"
-            )
-            assert isinstance(resource_response[0], TextResourceContents)
-            assert resource_response[0].text == "pong"
-
 
 class TestResourceTemplates:
     async def test_list_resource_templates(
@@ -301,8 +247,7 @@ class TestResourceTemplates:
         """
         By default, resource templates exclude GET methods without parameters
         """
-        async with Client(fastmcp_openapi_server) as client:
-            resource_templates = await client.list_resource_templates()
+        resource_templates = await fastmcp_openapi_server._mcp_list_resource_templates()
         assert len(resource_templates) == 1
         assert resource_templates[0].name == "get_user_users__user_id__get"
         assert (
@@ -320,14 +265,11 @@ class TestResourceTemplates:
         The resource template created by the OpenAPI server should be the same as the original
         """
         user_id = 2
-        async with Client(fastmcp_openapi_server) as client:
-            resource_response = await client.read_resource(
-                f"resource://openapi/get_user_users__user_id__get/{user_id}"
-            )
-            assert isinstance(resource_response[0], TextResourceContents)
-            response_text = resource_response[0].text
-            resource = json.loads(response_text)
+        resource_response = await fastmcp_openapi_server._mcp_read_resource(
+            f"resource://openapi/get_user_users__user_id__get/{user_id}"
+        )
 
+        resource = resource_response[0].content
         assert resource == users_db[user_id].model_dump()
         response = await api_client.get(f"/users/{user_id}")
         assert resource == response.json()
@@ -338,8 +280,7 @@ class TestPrompts:
         """
         By default, there are no prompts.
         """
-        async with Client(fastmcp_openapi_server) as client:
-            prompts = await client.list_prompts()
+        prompts = await fastmcp_openapi_server._mcp_list_prompts()
         assert len(prompts) == 0
 
 
@@ -553,23 +494,20 @@ class TestOpenAPI30Compatibility:
 
     async def test_resource_discovery(self, openapi_30_server):
         """Test that resources are correctly discovered from an OpenAPI 3.0 spec."""
-        async with Client(openapi_30_server) as client:
-            resources = await client.list_resources()
+        resources = await openapi_30_server._mcp_list_resources()
         assert len(resources) == 1
         assert resources[0].uri == AnyUrl("resource://openapi/listProducts")
 
     async def test_resource_template_discovery(self, openapi_30_server):
         """Test that resource templates are correctly discovered from an OpenAPI 3.0 spec."""
-        async with Client(openapi_30_server) as client:
-            templates = await client.list_resource_templates()
+        templates = await openapi_30_server._mcp_list_resource_templates()
         assert len(templates) == 1
         assert templates[0].name == "getProduct"
         assert templates[0].uriTemplate == r"resource://openapi/getProduct/{product_id}"
 
     async def test_tool_discovery(self, openapi_30_server):
         """Test that tools are correctly discovered from an OpenAPI 3.0 spec."""
-        async with Client(openapi_30_server) as client:
-            tools = await client.list_tools()
+        tools = await openapi_30_server._mcp_list_tools()
         assert len(tools) == 1
         assert tools[0].name == "createProduct"
         assert "name" in tools[0].inputSchema["properties"]
@@ -577,26 +515,20 @@ class TestOpenAPI30Compatibility:
 
     async def test_resource_access(self, openapi_30_server):
         """Test reading a resource from an OpenAPI 3.0 server."""
-        async with Client(openapi_30_server) as client:
-            resource_response = await client.read_resource(
-                "resource://openapi/listProducts"
-            )
-            assert isinstance(resource_response[0], TextResourceContents)
-            response_text = resource_response[0].text
-            content = json.loads(response_text)
+        resource_response = await openapi_30_server._mcp_read_resource(
+            "resource://openapi/listProducts"
+        )
+        content = resource_response[0].content
         assert len(content) == 2
         assert content[0]["name"] == "Product 1"
         assert content[1]["name"] == "Product 2"
 
     async def test_resource_template_access(self, openapi_30_server):
         """Test reading a resource from template from an OpenAPI 3.0 server."""
-        async with Client(openapi_30_server) as client:
-            resource_response = await client.read_resource(
-                "resource://openapi/getProduct/p1"
-            )
-            assert isinstance(resource_response[0], TextResourceContents)
-            response_text = resource_response[0].text
-            content = json.loads(response_text)
+        resource_response = await openapi_30_server._mcp_read_resource(
+            "resource://openapi/getProduct/p1"
+        )
+        content = resource_response[0].content
         assert content["id"] == "p1"
         assert content["name"] == "Product 1"
         assert content["price"] == 19.99
@@ -733,23 +665,20 @@ class TestOpenAPI31Compatibility:
 
     async def test_resource_discovery(self, openapi_31_server):
         """Test that resources are correctly discovered from an OpenAPI 3.1 spec."""
-        async with Client(openapi_31_server) as client:
-            resources = await client.list_resources()
+        resources = await openapi_31_server._mcp_list_resources()
         assert len(resources) == 1
         assert resources[0].uri == AnyUrl("resource://openapi/listOrders")
 
     async def test_resource_template_discovery(self, openapi_31_server):
         """Test that resource templates are correctly discovered from an OpenAPI 3.1 spec."""
-        async with Client(openapi_31_server) as client:
-            templates = await client.list_resource_templates()
+        templates = await openapi_31_server._mcp_list_resource_templates()
         assert len(templates) == 1
         assert templates[0].name == "getOrder"
         assert templates[0].uriTemplate == r"resource://openapi/getOrder/{order_id}"
 
     async def test_tool_discovery(self, openapi_31_server):
         """Test that tools are correctly discovered from an OpenAPI 3.1 spec."""
-        async with Client(openapi_31_server) as client:
-            tools = await client.list_tools()
+        tools = await openapi_31_server._mcp_list_tools()
         assert len(tools) == 1
         assert tools[0].name == "createOrder"
         assert "customer" in tools[0].inputSchema["properties"]
@@ -757,26 +686,20 @@ class TestOpenAPI31Compatibility:
 
     async def test_resource_access(self, openapi_31_server):
         """Test reading a resource from an OpenAPI 3.1 server."""
-        async with Client(openapi_31_server) as client:
-            resource_response = await client.read_resource(
-                "resource://openapi/listOrders"
-            )
-            assert isinstance(resource_response[0], TextResourceContents)
-            response_text = resource_response[0].text
-            content = json.loads(response_text)
+        resource_response = await openapi_31_server._mcp_read_resource(
+            "resource://openapi/listOrders"
+        )
+        content = resource_response[0].content
         assert len(content) == 2
         assert content[0]["customer"] == "Alice"
         assert content[1]["customer"] == "Bob"
 
     async def test_resource_template_access(self, openapi_31_server):
         """Test reading a resource from template from an OpenAPI 3.1 server."""
-        async with Client(openapi_31_server) as client:
-            resource_response = await client.read_resource(
-                "resource://openapi/getOrder/o1"
-            )
-            assert isinstance(resource_response[0], TextResourceContents)
-            response_text = resource_response[0].text
-            content = json.loads(response_text)
+        resource_response = await openapi_31_server._mcp_read_resource(
+            "resource://openapi/getOrder/o1"
+        )
+        content = resource_response[0].content
         assert content["id"] == "o1"
         assert content["customer"] == "Alice"
         assert content["items"] == ["item1", "item2"]
@@ -806,9 +729,8 @@ class TestMountFastMCP:
         await mcp.import_server("fastapi", fastmcp_openapi_server)
 
         # Check that resources are available with prefixed URIs
-        async with Client(mcp) as client:
-            resources = await client.list_resources()
-        assert len(resources) == 3
+        resources = await mcp._mcp_list_resources()
+        assert len(resources) == 1
         # We're checking the key used by mcp to store the resource
         # The prefixed URI is used as the key, but the resource's original uri is preserved
         prefixed_uri = "fastapi+resource://openapi/get_users_users_get"
@@ -816,8 +738,7 @@ class TestMountFastMCP:
         assert resource is not None
 
         # Check that templates are available with prefixed URIs
-        async with Client(mcp) as client:
-            templates = await client.list_resource_templates()
+        templates = await mcp._mcp_list_resource_templates()
         assert len(templates) == 1
         assert templates[0].name == "get_user_users__user_id__get"
         prefixed_template_uri = (
@@ -827,12 +748,10 @@ class TestMountFastMCP:
         assert template is not None
 
         # Check that tools are available with prefixed names
-        async with Client(mcp) as client:
-            tools = await client.list_tools()
+        tools = await mcp._mcp_list_tools()
         assert len(tools) == 2
         assert tools[0].name == "fastapi_create_user_users_post"
         assert tools[1].name == "fastapi_update_user_name_users__user_id__name_patch"
 
-        async with Client(mcp) as client:
-            prompts = await client.list_prompts()
+        prompts = await mcp._mcp_list_prompts()
         assert len(prompts) == 0