Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
742f56dc52 |
+34
-4
@@ -1,6 +1,7 @@
|
||||
import json
|
||||
from mcp.server import Server
|
||||
from mcp.types import Resource, TextContent
|
||||
from mcp.server.lowlevel.helper_types import ReadResourceContents
|
||||
from mcp.types import Resource, ResourceTemplate
|
||||
from .backend_client import backend
|
||||
|
||||
RESOURCE_LIST = [
|
||||
@@ -19,8 +20,26 @@ ROUTES = {
|
||||
"homelable://scan/runs": "/api/v1/scan/runs",
|
||||
}
|
||||
|
||||
# read_resource() also serves homelable://nodes/<id>, which is not in
|
||||
# RESOURCE_LIST because it is a template, not a concrete URI. Without a
|
||||
# list_resource_templates handler the SDK answers resources/templates/list with
|
||||
# "Method not found" and the template stays invisible to every client.
|
||||
RESOURCE_TEMPLATES = [
|
||||
ResourceTemplate(
|
||||
uriTemplate="homelable://nodes/{node_id}",
|
||||
name="Node",
|
||||
description="A single node by id",
|
||||
mimeType="application/json",
|
||||
),
|
||||
]
|
||||
|
||||
async def read_resource(uri: str) -> list[TextContent]:
|
||||
|
||||
# The low-level Server.read_resource decorator expects str, bytes or an
|
||||
# iterable of ReadResourceContents, and reads .content / .mime_type off each
|
||||
# item. Returning mcp.types.TextContent instead fails at serialisation time with
|
||||
# "'TextContent' object has no attribute 'content'" — the handler runs, so the
|
||||
# error only surfaces to the client, never in the server log.
|
||||
async def read_resource(uri: str) -> list[ReadResourceContents]:
|
||||
# The MCP framework hands us a pydantic AnyUrl, not a plain str, so string
|
||||
# ops like .startswith / dict lookups blow up with
|
||||
# "'AnyUrl' object has no attribute 'startswith'". Coerce to str first.
|
||||
@@ -28,13 +47,20 @@ async def read_resource(uri: str) -> list[TextContent]:
|
||||
if uri.startswith("homelable://nodes/") and uri != "homelable://nodes/":
|
||||
node_id = uri.split("/")[-1]
|
||||
data = await backend.get(f"/api/v1/nodes/{node_id}")
|
||||
return [TextContent(type="text", text=json.dumps(data, indent=2))]
|
||||
return [_json_contents(data)]
|
||||
|
||||
if uri not in ROUTES:
|
||||
raise ValueError(f"Unknown resource URI: {uri}")
|
||||
|
||||
data = await backend.get(ROUTES[uri])
|
||||
return [TextContent(type="text", text=json.dumps(data, indent=2))]
|
||||
return [_json_contents(data)]
|
||||
|
||||
|
||||
def _json_contents(data) -> ReadResourceContents:
|
||||
return ReadResourceContents(
|
||||
content=json.dumps(data, indent=2),
|
||||
mime_type="application/json",
|
||||
)
|
||||
|
||||
|
||||
def register_resources(server: Server):
|
||||
@@ -42,6 +68,10 @@ def register_resources(server: Server):
|
||||
async def _list():
|
||||
return RESOURCE_LIST
|
||||
|
||||
@server.list_resource_templates()
|
||||
async def _list_templates():
|
||||
return RESOURCE_TEMPLATES
|
||||
|
||||
@server.read_resource()
|
||||
async def _read(uri: str):
|
||||
return await read_resource(uri)
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import json
|
||||
import pytest
|
||||
from pydantic import AnyUrl
|
||||
from unittest.mock import AsyncMock, patch
|
||||
@@ -67,3 +68,46 @@ async def test_read_edges_anyurl(mock_backend):
|
||||
async def test_read_single_node_anyurl(mock_backend):
|
||||
await read_resource(AnyUrl("homelable://nodes/abc123"))
|
||||
mock_backend.get.assert_called_once_with("/api/v1/nodes/abc123")
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_read_returns_read_resource_contents(mock_backend):
|
||||
"""Regression: returning mcp.types.TextContent made every resources/read
|
||||
fail client-side with "'TextContent' object has no attribute 'content'".
|
||||
The low-level Server reads .content / .mime_type off each item."""
|
||||
from mcp.server.lowlevel.helper_types import ReadResourceContents
|
||||
|
||||
result = await read_resource("homelable://canvas")
|
||||
|
||||
assert isinstance(result[0], ReadResourceContents)
|
||||
assert result[0].mime_type == "application/json"
|
||||
assert json.loads(result[0].content) == {"data": "ok"}
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_read_node_returns_read_resource_contents(mock_backend):
|
||||
from mcp.server.lowlevel.helper_types import ReadResourceContents
|
||||
|
||||
result = await read_resource("homelable://nodes/abc123")
|
||||
|
||||
assert isinstance(result[0], ReadResourceContents)
|
||||
assert result[0].mime_type == "application/json"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_resource_templates_are_registered():
|
||||
"""Regression: no list_resource_templates handler meant
|
||||
resources/templates/list answered "Method not found", hiding the
|
||||
homelable://nodes/{node_id} template read_resource already serves."""
|
||||
from mcp.server import Server
|
||||
from mcp.types import ListResourceTemplatesRequest
|
||||
from app.resources import register_resources
|
||||
|
||||
server = Server("test")
|
||||
register_resources(server)
|
||||
|
||||
handler = server.request_handlers[ListResourceTemplatesRequest]
|
||||
result = await handler(None)
|
||||
|
||||
templates = result.root.resourceTemplates
|
||||
assert [t.uriTemplate for t in templates] == ["homelable://nodes/{node_id}"]
|
||||
|
||||
Reference in New Issue
Block a user