mirror of
https://github.com/aljazceru/mcp-python-sdk.git
synced 2025-12-19 06:54:18 +01:00
b1942b31c4
* feat: allow lowlevel servers to return a list of resources The resource/read message in MCP allows of multiple resources to be returned. However, in the SDK we do not allow this. This change is such that we allow returning multiple resource in the lowlevel API if needed. However in FastMCP we stick to one, since a FastMCP resource defines the mime_type in the decorator and hence a resource cannot dynamically return different mime_typed resources. It also is just the better default to only return one resource. However in the lowlevel API we will allow this. Strictly speaking this is not a BC break since the new return value is additive, but if people subclassed server, it will break them. * feat: lower the type requriements for call_tool to Iterable
133 lines
3.5 KiB
Python
133 lines
3.5 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
|
|
@pytest.fixture()
|
|
def test_dir(tmp_path_factory) -> Path:
|
|
"""Create a temporary directory with test files."""
|
|
tmp = tmp_path_factory.mktemp("test_files")
|
|
|
|
# Create test files
|
|
(tmp / "example.py").write_text("print('hello world')")
|
|
(tmp / "readme.md").write_text("# Test Directory\nThis is a test.")
|
|
(tmp / "config.json").write_text('{"test": true}')
|
|
|
|
return tmp
|
|
|
|
|
|
@pytest.fixture
|
|
def mcp() -> FastMCP:
|
|
mcp = FastMCP()
|
|
|
|
return mcp
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def resources(mcp: FastMCP, test_dir: Path) -> FastMCP:
|
|
@mcp.resource("dir://test_dir")
|
|
def list_test_dir() -> list[str]:
|
|
"""List the files in the test directory"""
|
|
return [str(f) for f in test_dir.iterdir()]
|
|
|
|
@mcp.resource("file://test_dir/example.py")
|
|
def read_example_py() -> str:
|
|
"""Read the example.py file"""
|
|
try:
|
|
return (test_dir / "example.py").read_text()
|
|
except FileNotFoundError:
|
|
return "File not found"
|
|
|
|
@mcp.resource("file://test_dir/readme.md")
|
|
def read_readme_md() -> str:
|
|
"""Read the readme.md file"""
|
|
try:
|
|
return (test_dir / "readme.md").read_text()
|
|
except FileNotFoundError:
|
|
return "File not found"
|
|
|
|
@mcp.resource("file://test_dir/config.json")
|
|
def read_config_json() -> str:
|
|
"""Read the config.json file"""
|
|
try:
|
|
return (test_dir / "config.json").read_text()
|
|
except FileNotFoundError:
|
|
return "File not found"
|
|
|
|
return mcp
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def tools(mcp: FastMCP, test_dir: Path) -> FastMCP:
|
|
@mcp.tool()
|
|
def delete_file(path: str) -> bool:
|
|
# ensure path is in test_dir
|
|
if Path(path).resolve().parent != test_dir:
|
|
raise ValueError(f"Path must be in test_dir: {path}")
|
|
Path(path).unlink()
|
|
return True
|
|
|
|
return mcp
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_list_resources(mcp: FastMCP):
|
|
resources = await mcp.list_resources()
|
|
assert len(resources) == 4
|
|
|
|
assert [str(r.uri) for r in resources] == [
|
|
"dir://test_dir",
|
|
"file://test_dir/example.py",
|
|
"file://test_dir/readme.md",
|
|
"file://test_dir/config.json",
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_read_resource_dir(mcp: FastMCP):
|
|
res_iter = await mcp.read_resource("dir://test_dir")
|
|
res_list = list(res_iter)
|
|
assert len(res_list) == 1
|
|
res = res_list[0]
|
|
assert res.mime_type == "text/plain"
|
|
|
|
files = json.loads(res.content)
|
|
|
|
assert sorted([Path(f).name for f in files]) == [
|
|
"config.json",
|
|
"example.py",
|
|
"readme.md",
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_read_resource_file(mcp: FastMCP):
|
|
res_iter = await mcp.read_resource("file://test_dir/example.py")
|
|
res_list = list(res_iter)
|
|
assert len(res_list) == 1
|
|
res = res_list[0]
|
|
assert res.content == "print('hello world')"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_delete_file(mcp: FastMCP, test_dir: Path):
|
|
await mcp.call_tool(
|
|
"delete_file", arguments=dict(path=str(test_dir / "example.py"))
|
|
)
|
|
assert not (test_dir / "example.py").exists()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_delete_file_and_check_resources(mcp: FastMCP, test_dir: Path):
|
|
await mcp.call_tool(
|
|
"delete_file", arguments=dict(path=str(test_dir / "example.py"))
|
|
)
|
|
res_iter = await mcp.read_resource("file://test_dir/example.py")
|
|
res_list = list(res_iter)
|
|
assert len(res_list) == 1
|
|
res = res_list[0]
|
|
assert res.content == "File not found"
|