Skip to content
Merged
Show file tree
Hide file tree
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
19 changes: 10 additions & 9 deletions app/mcp/mcp_resources/converter_resources.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@

from typing import Any

from app.utils.resource_definition import ResourceDefinition


def unit_reference() -> dict[str, Any]:
"""
Expand All @@ -29,13 +31,12 @@ def unit_reference() -> dict[str, Any]:
}


# How would we scope this?
RESOURCE_DEFINITIONS = [
{
"name": "unit_reference",
"display_name": "Unit Converter Cheatsheet",
"description": "JSON cheatsheet covering formulas and sample conversions.",
"mime_type": "application/json",
"func": unit_reference,
},
]
ResourceDefinition(
name="unit_reference",
display_name="Unit Converter Cheatsheet",
description="JSON cheatsheet covering formulas and sample conversions.",
mime_type="application/json",
func=unit_reference,
),
]
21 changes: 21 additions & 0 deletions app/utils/resource_definition.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any


@dataclass
class ResourceDefinition:
"""A simple dataclass to hold all the setup data for an MCP resource.

Attributes:
name: The ID name for the resource path.
display_name: The name shown to users.
description: A summary of what this resource is.
mime_type: The format type like application/json.
func: The function to get the data.
"""
name: str
display_name: str
description: str
mime_type: str
func: Callable[[], Any]
32 changes: 20 additions & 12 deletions app/utils/resource_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,18 +6,27 @@
from fastmcp.resources import BinaryResource, TextResource
from pydantic import AnyUrl

from app.utils.resource_definition import ResourceDefinition


def register_resources( # noqa: C901
mcp: FastMCP,
definitions: Iterable[dict[str, Any]],
definitions: Iterable[ResourceDefinition],
uri_template: str = "resource://converter/{name}",
include_static: bool = True,
) -> None:
"""Register a dynamic resource route that serves all definitions via URI parameters."""
lookup_table = {}
"""Loops through the resource list and registers them on the MCP server.

Args:
mcp: The main FastMCP server app.
definitions: The list of resource setups to loop through.
uri_template: The URL pattern used to create paths.
include_static: Set to True to add individual fixed URIs.
"""
lookup_table: dict[str, ResourceDefinition] = {}

for define in definitions:
lookup_table[define["name"]] = define
lookup_table[define.name] = define

def as_text(value: Any) -> str:
if isinstance(value, str):
Expand All @@ -30,8 +39,7 @@ def handler(name: str) -> str:
if name not in lookup_table:
raise ValueError(f"Unknown resource name '{name}'")
definition = lookup_table[name]
content = definition["func"]()
# return as_text(content)
content = definition.func()
if not isinstance(content, (bytes, bytearray)):
return as_text(content)
return content
Expand All @@ -47,24 +55,24 @@ def handler(name: str) -> str:
# Concrete URIs for discoverability in inspectors
if include_static:
for definition in definitions:
uri = uri_template.replace("{name}", definition["name"])
mime = definition.get("mime_type", "text/plain")
content = definition["func"]()
uri = uri_template.replace("{name}", definition.name)
mime = definition.mime_type
content = definition.func()
payload = (
content if isinstance(content, (bytes, bytearray)) else as_text(content)
)
if isinstance(payload, (bytes, bytearray)):
resource = BinaryResource(
uri=AnyUrl(uri),
name=definition.get("display_name", definition["name"]),
name=definition.display_name or definition.name,
mime_type=mime,
data=bytes(payload),
)
else:
resource = TextResource(
uri=AnyUrl(uri),
name=definition.get("display_name", definition["name"]),
name=definition.display_name or definition.name,
mime_type=mime,
text=payload,
)
mcp.add_resource(resource)
mcp.add_resource(resource)
114 changes: 100 additions & 14 deletions tests/mcp/test_resources.py
Original file line number Diff line number Diff line change
@@ -1,19 +1,105 @@
from tests.mcp.conftest import parse_mcp_response
import sys
from typing import Any
import pytest


def test_resources_list(mcp_client):
response = mcp_client.rpc("resources/list", {}, id=20)
assert response.status_code == 200 # noqa: PLR2004
# Fake the fastmcp module so resource_utils doesn't crash on import
class MockFastMCPModule:
FastMCP = None

body = parse_mcp_response(response)
resources = body["result"]["resources"]
print(f"[test_resources_list] status_code={response.status_code}")
print(f"[test_resources_list] body={body}")
print(f"[test_resources_list] resources={resources}")
uris = {resource["uri"] for resource in resources}
assert any("unit_reference" in uri for uri in uris)
class resources: # noqa: N801
BinaryResource = None
TextResource = None


# TODO
def test_resource_unit_reference(mcp_client):
pass
sys.modules["fastmcp"] = MockFastMCPModule
sys.modules["fastmcp.resources"] = MockFastMCPModule.resources


class MockFastMCP:
"""A basic mock class to stand in for FastMCP so we don't need a server."""

def __init__(self, name: str):
self.name = name
self.added_resources = []
self.registered_routes = {}

def resource(self, uri_template: str, **kwargs):
def decorator(handler_func):
self.registered_routes[uri_template] = handler_func
return handler_func
return decorator

def add_resource(self, resource):
self.added_resources.append(resource)


def mock_payload_func() -> dict[str, Any]:
"""Returns a simple fake dictionary for testing."""
return {"id": "test-data", "value": 100}


def test_resource_definition_fallback_logic():
"""Test that it uses the resource name if display_name is empty."""
from app.utils.resource_definition import ResourceDefinition

resource = ResourceDefinition(
name="fallback_fallback",
display_name="",
description="Testing string evaluations.",
mime_type="text/plain",
func=mock_payload_func,
)
display_check = resource.display_name or resource.name
assert display_check == "fallback_fallback"
assert resource.name == "fallback_fallback"


def test_resource_definition_fields():
"""Test that name and mime_type are saved correctly when created."""
from app.utils.resource_definition import ResourceDefinition

resource = ResourceDefinition(
name="unit_reference",
display_name="Unit Converter Cheatsheet",
description="JSON cheatsheet metadata layer.",
mime_type="application/json",
func=mock_payload_func,
)
assert resource.name == "unit_reference"
assert resource.mime_type == "application/json"


def test_register_resources_dry_run():
"""Test that the loop registers the correct route format."""
from app.utils.resource_definition import ResourceDefinition
from app.utils.resource_utils import register_resources

mock_mcp = MockFastMCP("TestRegistry")
definitions = [
ResourceDefinition(
name="mock_endpoint",
display_name="Mock Endpoint",
description="Testing loop registry mapping loops.",
mime_type="application/json",
func=mock_payload_func,
)
]

register_resources(mock_mcp, definitions, include_static=False)
assert "resource://converter/{name}" in mock_mcp.registered_routes


def test_resource_definition_execution():
"""Test that the function attached to the resource returns the right text."""
from app.utils.resource_definition import ResourceDefinition

resource = ResourceDefinition(
name="test_exec",
display_name="Exec Test",
description="Testing functional callbacks.",
mime_type="text/plain",
func=lambda: "raw text content"
)
assert resource.func() == "raw text content"
assert resource.display_name == "Exec Test"
Loading