mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-24 06:24:18 +02:00
851 lines
31 KiB
Python
851 lines
31 KiB
Python
import base64
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from mcp.types import (
|
|
BlobResourceContents,
|
|
ImageContent,
|
|
TextContent,
|
|
TextResourceContents,
|
|
)
|
|
from pydantic import AnyUrl, Field
|
|
|
|
from fastmcp import Client, Context, FastMCP
|
|
from fastmcp.exceptions import ClientError
|
|
from fastmcp.prompts.prompt import EmbeddedResource, Message, UserMessage
|
|
from fastmcp.resources import FileResource, FunctionResource
|
|
from fastmcp.utilities.types import Image
|
|
|
|
|
|
@pytest.fixture
|
|
def tool_server():
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool()
|
|
def add(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
@mcp.tool()
|
|
def list_tool() -> list[str | int]:
|
|
return ["x", 2]
|
|
|
|
@mcp.tool()
|
|
def error_tool() -> None:
|
|
raise ValueError("Test error")
|
|
|
|
@mcp.tool()
|
|
def image_tool(path: str) -> Image:
|
|
return Image(path)
|
|
|
|
@mcp.tool()
|
|
def mixed_content_tool() -> list[TextContent | ImageContent]:
|
|
return [
|
|
TextContent(type="text", text="Hello"),
|
|
ImageContent(type="image", data="abc", mimeType="image/png"),
|
|
]
|
|
|
|
@mcp.tool()
|
|
def mixed_list_fn(image_path: str) -> list:
|
|
return [
|
|
"text message",
|
|
Image(image_path),
|
|
{"key": "value"},
|
|
TextContent(type="text", text="direct content"),
|
|
]
|
|
|
|
return mcp
|
|
|
|
|
|
class TestTools:
|
|
async def test_add_tool_exists(self, tool_server: FastMCP):
|
|
async with Client(tool_server) as client:
|
|
tools = await client.list_tools()
|
|
assert "add" in [t.name for t in tools]
|
|
|
|
async def test_list_tools(self, tool_server: FastMCP):
|
|
async with Client(tool_server) as client:
|
|
assert len(await client.list_tools()) == 6
|
|
|
|
async def test_call_tool(self, tool_server: FastMCP):
|
|
async with Client(tool_server) as client:
|
|
result = await client.call_tool("add", {"x": 1, "y": 2})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == "3"
|
|
|
|
async def test_call_tool_as_client(self, tool_server: FastMCP):
|
|
async with Client(tool_server) as client:
|
|
result = await client.call_tool("add", {"x": 1, "y": 2})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == "3"
|
|
|
|
async def test_call_tool_error(self, tool_server: FastMCP):
|
|
async with Client(tool_server) as client:
|
|
with pytest.raises(Exception):
|
|
await client.call_tool("error_tool", {})
|
|
|
|
async def test_call_tool_error_as_client(self, tool_server: FastMCP):
|
|
async with Client(tool_server) as client:
|
|
with pytest.raises(Exception):
|
|
await client.call_tool("error_tool", {})
|
|
|
|
async def test_call_tool_error_as_client_raw(self, tool_server: FastMCP):
|
|
async with Client(tool_server) as client:
|
|
result = await client.call_tool("error_tool", {}, _return_raw_result=True)
|
|
assert result.isError
|
|
assert isinstance(result.content[0], TextContent)
|
|
assert "Test error" in result.content[0].text
|
|
|
|
async def test_tool_returns_list(self, tool_server: FastMCP):
|
|
async with Client(tool_server) as client:
|
|
result = await client.call_tool("list_tool", {})
|
|
assert isinstance(result[0], TextContent)
|
|
assert result[0].text == '["x", 2]'
|
|
|
|
async def test_tool_image_helper(self, tool_server: FastMCP, tmp_path: Path):
|
|
# Create a test image
|
|
image_path = tmp_path / "test.png"
|
|
image_path.write_bytes(b"fake png data")
|
|
|
|
async with Client(tool_server) as client:
|
|
result = await client.call_tool("image_tool", {"path": str(image_path)})
|
|
content = result[0]
|
|
assert isinstance(content, ImageContent)
|
|
assert content.type == "image"
|
|
assert content.mimeType == "image/png"
|
|
# Verify base64 encoding
|
|
decoded = base64.b64decode(content.data)
|
|
assert decoded == b"fake png data"
|
|
|
|
async def test_tool_mixed_content(self, tool_server: FastMCP):
|
|
async with Client(tool_server) as client:
|
|
result = await client.call_tool("mixed_content_tool", {})
|
|
assert len(result) == 2
|
|
content1 = result[0]
|
|
content2 = result[1]
|
|
assert isinstance(content1, TextContent)
|
|
assert content1.text == "Hello"
|
|
assert isinstance(content2, ImageContent)
|
|
assert content2.mimeType == "image/png"
|
|
assert content2.data == "abc"
|
|
|
|
async def test_tool_mixed_list_with_image(
|
|
self, tool_server: FastMCP, tmp_path: Path
|
|
):
|
|
"""Test that lists containing Image objects and other types are handled
|
|
correctly. Note that the non-MCP content will be grouped together."""
|
|
# Create a test image
|
|
image_path = tmp_path / "test.png"
|
|
image_path.write_bytes(b"test image data")
|
|
|
|
async with Client(tool_server) as client:
|
|
result = await client.call_tool(
|
|
"mixed_list_fn", {"image_path": str(image_path)}
|
|
)
|
|
assert len(result) == 3
|
|
# Check text conversion
|
|
content1 = result[0]
|
|
assert isinstance(content1, TextContent)
|
|
assert json.loads(content1.text) == ["text message", {"key": "value"}]
|
|
# Check image conversion
|
|
content2 = result[1]
|
|
assert isinstance(content2, ImageContent)
|
|
assert content2.mimeType == "image/png"
|
|
assert base64.b64decode(content2.data) == b"test image data"
|
|
# Check direct TextContent
|
|
content3 = result[2]
|
|
assert isinstance(content3, TextContent)
|
|
assert content3.text == "direct content"
|
|
|
|
async def test_parameter_descriptions(self):
|
|
mcp = FastMCP("Test Server")
|
|
|
|
@mcp.tool()
|
|
def greet(
|
|
name: str = Field(description="The name to greet"),
|
|
title: str = Field(description="Optional title", default=""),
|
|
) -> str:
|
|
"""A greeting tool"""
|
|
return f"Hello {title} {name}"
|
|
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
assert len(tools) == 1
|
|
tool = tools[0]
|
|
|
|
# Check that parameter descriptions are present in the schema
|
|
properties = tool.inputSchema["properties"]
|
|
assert "name" in properties
|
|
assert properties["name"]["description"] == "The name to greet"
|
|
assert "title" in properties
|
|
assert properties["title"]["description"] == "Optional title"
|
|
|
|
|
|
class TestResources:
|
|
async def test_text_resource(self):
|
|
mcp = FastMCP()
|
|
|
|
def get_text():
|
|
return "Hello, world!"
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("resource://test"), name="test", fn=get_text
|
|
)
|
|
mcp.add_resource(resource)
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(AnyUrl("resource://test"))
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Hello, world!"
|
|
|
|
async def test_binary_resource(self):
|
|
mcp = FastMCP()
|
|
|
|
def get_binary():
|
|
return b"Binary data"
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("resource://binary"),
|
|
name="binary",
|
|
fn=get_binary,
|
|
mime_type="application/octet-stream",
|
|
)
|
|
mcp.add_resource(resource)
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(AnyUrl("resource://binary"))
|
|
assert isinstance(result[0], BlobResourceContents)
|
|
assert result[0].blob == base64.b64encode(b"Binary data").decode()
|
|
|
|
async def test_file_resource_text(self, tmp_path: Path):
|
|
mcp = FastMCP()
|
|
|
|
# Create a text file
|
|
text_file = tmp_path / "test.txt"
|
|
text_file.write_text("Hello from file!")
|
|
|
|
resource = FileResource(
|
|
uri=AnyUrl("file://test.txt"), name="test.txt", path=text_file
|
|
)
|
|
mcp.add_resource(resource)
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(AnyUrl("file://test.txt"))
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Hello from file!"
|
|
|
|
async def test_file_resource_binary(self, tmp_path: Path):
|
|
mcp = FastMCP()
|
|
|
|
# Create a binary file
|
|
binary_file = tmp_path / "test.bin"
|
|
binary_file.write_bytes(b"Binary file data")
|
|
|
|
resource = FileResource(
|
|
uri=AnyUrl("file://test.bin"),
|
|
name="test.bin",
|
|
path=binary_file,
|
|
mime_type="application/octet-stream",
|
|
)
|
|
mcp.add_resource(resource)
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(AnyUrl("file://test.bin"))
|
|
assert isinstance(result[0], BlobResourceContents)
|
|
assert result[0].blob == base64.b64encode(b"Binary file data").decode()
|
|
|
|
|
|
class TestResourceTemplates:
|
|
async def test_resource_with_params_not_in_uri(self):
|
|
"""Test that a resource with function parameters raises an error if the URI
|
|
parameters don't match"""
|
|
mcp = FastMCP()
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="URI template must contain at least one parameter",
|
|
):
|
|
|
|
@mcp.resource("resource://data")
|
|
def get_data_fn(param: str) -> str:
|
|
return f"Data: {param}"
|
|
|
|
async def test_resource_with_uri_params_without_args(self):
|
|
"""Test that a resource with URI parameters is automatically a template"""
|
|
mcp = FastMCP()
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="URI parameters .* must be a subset of the function arguments",
|
|
):
|
|
|
|
@mcp.resource("resource://{param}")
|
|
def get_data() -> str:
|
|
return "Data"
|
|
|
|
async def test_resource_with_untyped_params(self):
|
|
"""Test that a resource with untyped parameters raises an error"""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{param}")
|
|
def get_data(param) -> str:
|
|
return "Data"
|
|
|
|
async def test_resource_matching_params(self):
|
|
"""Test that a resource with matching URI and function parameters works"""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{name}/data")
|
|
def get_data(name: str) -> str:
|
|
return f"Data for {name}"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(AnyUrl("resource://test/data"))
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Data for test"
|
|
|
|
async def test_resource_mismatched_params(self):
|
|
"""Test that mismatched parameters raise an error"""
|
|
mcp = FastMCP()
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="URI parameters .* must be a subset of the required function arguments",
|
|
):
|
|
|
|
@mcp.resource("resource://{name}/data")
|
|
def get_data(user: str) -> str:
|
|
return f"Data for {user}"
|
|
|
|
async def test_resource_multiple_params(self):
|
|
"""Test that multiple parameters work correctly"""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{org}/{repo}/data")
|
|
def get_data(org: str, repo: str) -> str:
|
|
return f"Data for {org}/{repo}"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(
|
|
AnyUrl("resource://cursor/fastmcp/data")
|
|
)
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Data for cursor/fastmcp"
|
|
|
|
async def test_resource_multiple_mismatched_params(self):
|
|
"""Test that mismatched parameters raise an error"""
|
|
mcp = FastMCP()
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="URI parameters .* must be a subset of the required function arguments",
|
|
):
|
|
|
|
@mcp.resource("resource://{org}/{repo}/data")
|
|
def get_data_mismatched(org: str, repo_2: str) -> str:
|
|
return f"Data for {org}"
|
|
|
|
"""Test that a resource with no parameters works as a regular resource"""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://static")
|
|
def get_static_data() -> str:
|
|
return "Static data"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(AnyUrl("resource://static"))
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Static data"
|
|
|
|
async def test_template_with_default_params(self):
|
|
"""Test that a template can have default parameters."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("math://add/{x}")
|
|
def add(x: int, y: int = 10) -> int:
|
|
return x + y
|
|
|
|
# Verify it's registered as a template
|
|
templates_dict = await mcp.get_resource_templates()
|
|
templates = list(templates_dict.values())
|
|
assert len(templates) == 1
|
|
assert templates[0].uri_template == "math://add/{x}"
|
|
|
|
# Call the template and verify it uses the default value
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(AnyUrl("math://add/5"))
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "15" # 5 + default 10
|
|
|
|
# Can also call with explicit params
|
|
result2 = await client.read_resource(AnyUrl("math://add/7"))
|
|
assert isinstance(result2[0], TextResourceContents)
|
|
assert result2[0].text == "17" # 7 + default 10
|
|
|
|
async def test_template_to_resource_conversion(self):
|
|
"""Test that a template can be converted to a resource."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{name}/data")
|
|
def get_data(name: str) -> str:
|
|
return f"Data for {name}"
|
|
|
|
# Verify it's registered as a template
|
|
templates_dict = await mcp.get_resource_templates()
|
|
templates = list(templates_dict.values())
|
|
assert len(templates) == 1
|
|
assert templates[0].uri_template == "resource://{name}/data"
|
|
|
|
# When accessed, should create a concrete resource
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(AnyUrl("resource://test/data"))
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Data for test"
|
|
|
|
async def test_stacked_resource_template_decorators(self):
|
|
"""Test that resource template decorators can be stacked."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("users://email/{email}")
|
|
@mcp.resource("users://name/{name}")
|
|
def lookup_user(name: str | None = None, email: str | None = None) -> dict:
|
|
if name:
|
|
return {
|
|
"id": "123",
|
|
"name": name,
|
|
"email": "dummy@example.com",
|
|
"lookup": "name",
|
|
}
|
|
elif email:
|
|
return {
|
|
"id": "123",
|
|
"name": "Test User",
|
|
"email": email,
|
|
"lookup": "email",
|
|
}
|
|
else:
|
|
raise ValueError("Either name or email must be provided")
|
|
|
|
# Verify both templates are registered
|
|
templates_dict = await mcp.get_resource_templates()
|
|
templates = list(templates_dict.values())
|
|
assert len(templates) == 2
|
|
template_uris = {t.uri_template for t in templates}
|
|
assert "users://email/{email}" in template_uris
|
|
assert "users://name/{name}" in template_uris
|
|
|
|
# Test lookup by email
|
|
async with Client(mcp) as client:
|
|
email_result = await client.read_resource(
|
|
AnyUrl("users://email/user@example.com")
|
|
)
|
|
assert isinstance(email_result[0], TextResourceContents)
|
|
email_data = json.loads(email_result[0].text)
|
|
assert email_data["lookup"] == "email"
|
|
assert email_data["email"] == "user@example.com"
|
|
|
|
# Test lookup by name
|
|
name_result = await client.read_resource(AnyUrl("users://name/John"))
|
|
assert isinstance(name_result[0], TextResourceContents)
|
|
name_data = json.loads(name_result[0].text)
|
|
assert name_data["lookup"] == "name"
|
|
assert name_data["name"] == "John"
|
|
assert name_data["email"] == "dummy@example.com"
|
|
|
|
async def test_template_decorator_with_tags(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{param}", tags={"template", "test-tag"})
|
|
def template_resource(param: str) -> str:
|
|
return f"Template resource: {param}"
|
|
|
|
templates_dict = await mcp.get_resource_templates()
|
|
template = templates_dict["resource://{param}"]
|
|
assert template.tags == {"template", "test-tag"}
|
|
|
|
async def test_template_decorator_wildcard_param(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{param*}")
|
|
def template_resource(param: str) -> str:
|
|
return f"Template resource: {param}"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(AnyUrl("resource://test/data"))
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Template resource: test/data"
|
|
|
|
async def test_templates_match_in_order_of_definition(self):
|
|
"""
|
|
If a wildcard template is defined first, it will take priority over another
|
|
matching template.
|
|
|
|
"""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{param*}")
|
|
def template_resource(param: str) -> str:
|
|
return f"Template resource 1: {param}"
|
|
|
|
@mcp.resource("resource://{x}/{y}")
|
|
def template_resource_with_params(x: str, y: str) -> str:
|
|
return f"Template resource 2: {x}/{y}"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(AnyUrl("resource://a/b/c"))
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Template resource 1: a/b/c"
|
|
|
|
result = await client.read_resource(AnyUrl("resource://a/b"))
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Template resource 1: a/b"
|
|
|
|
async def test_templates_shadow_each_other_reorder(self):
|
|
"""
|
|
If a wildcard template is defined second, it will *not* take priority over
|
|
another matching template.
|
|
"""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{x}/{y}")
|
|
def template_resource_with_params(x: str, y: str) -> str:
|
|
return f"Template resource 1: {x}/{y}"
|
|
|
|
@mcp.resource("resource://{param*}")
|
|
def template_resource(param: str) -> str:
|
|
return f"Template resource 2: {param}"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(AnyUrl("resource://a/b/c"))
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Template resource 2: a/b/c"
|
|
|
|
result = await client.read_resource(AnyUrl("resource://a/b"))
|
|
assert isinstance(result[0], TextResourceContents)
|
|
assert result[0].text == "Template resource 1: a/b"
|
|
|
|
|
|
class TestContextInjection:
|
|
"""Test context injection in tools."""
|
|
|
|
async def test_context_detection(self):
|
|
"""Test that context parameters are properly detected."""
|
|
mcp = FastMCP()
|
|
|
|
def tool_with_context(x: int, ctx: Context) -> str:
|
|
return f"Request {ctx.request_id}: {x}"
|
|
|
|
mcp.add_tool(tool_with_context)
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "tool_with_context"
|
|
|
|
async def test_context_injection(self):
|
|
"""Test that context is properly injected into tool calls."""
|
|
mcp = FastMCP()
|
|
|
|
def tool_with_context(x: int, ctx: Context) -> str:
|
|
assert ctx.request_id is not None
|
|
return f"Request {ctx.request_id}: {x}"
|
|
|
|
mcp.add_tool(tool_with_context)
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("tool_with_context", {"x": 42})
|
|
assert len(result) == 1
|
|
content = result[0]
|
|
assert isinstance(content, TextContent)
|
|
assert "Request" in content.text
|
|
assert "42" in content.text
|
|
|
|
async def test_async_context(self):
|
|
"""Test that context works in async functions."""
|
|
mcp = FastMCP()
|
|
|
|
async def async_tool(x: int, ctx: Context) -> str:
|
|
assert ctx.request_id is not None
|
|
return f"Async request {ctx.request_id}: {x}"
|
|
|
|
mcp.add_tool(async_tool)
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("async_tool", {"x": 42})
|
|
assert len(result) == 1
|
|
content = result[0]
|
|
assert isinstance(content, TextContent)
|
|
assert "Async request" in content.text
|
|
assert "42" in content.text
|
|
|
|
async def test_context_logging(self):
|
|
from unittest.mock import patch
|
|
|
|
import mcp.server.session
|
|
|
|
"""Test that context logging methods work."""
|
|
mcp = FastMCP()
|
|
|
|
async def logging_tool(msg: str, ctx: Context) -> str:
|
|
await ctx.debug("Debug message")
|
|
await ctx.info("Info message")
|
|
await ctx.warning("Warning message")
|
|
await ctx.error("Error message")
|
|
return f"Logged messages for {msg}"
|
|
|
|
mcp.add_tool(logging_tool)
|
|
|
|
with patch("mcp.server.session.ServerSession.send_log_message") as mock_log:
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("logging_tool", {"msg": "test"})
|
|
assert len(result) == 1
|
|
content = result[0]
|
|
assert isinstance(content, TextContent)
|
|
assert "Logged messages for test" in content.text
|
|
|
|
assert mock_log.call_count == 4
|
|
mock_log.assert_any_call(
|
|
level="debug", data="Debug message", logger=None
|
|
)
|
|
mock_log.assert_any_call(level="info", data="Info message", logger=None)
|
|
mock_log.assert_any_call(
|
|
level="warning", data="Warning message", logger=None
|
|
)
|
|
mock_log.assert_any_call(
|
|
level="error", data="Error message", logger=None
|
|
)
|
|
|
|
async def test_optional_context(self):
|
|
"""Test that context is optional."""
|
|
mcp = FastMCP()
|
|
|
|
def no_context(x: int) -> int:
|
|
return x * 2
|
|
|
|
mcp.add_tool(no_context)
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("no_context", {"x": 21})
|
|
assert len(result) == 1
|
|
content = result[0]
|
|
assert isinstance(content, TextContent)
|
|
assert content.text == "42"
|
|
|
|
async def test_context_resource_access(self):
|
|
"""Test that context can access resources."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("test://data")
|
|
def test_resource() -> str:
|
|
return "resource data"
|
|
|
|
@mcp.tool()
|
|
async def tool_with_resource(ctx: Context) -> str:
|
|
r_iter = await ctx.read_resource("test://data")
|
|
r_list = list(r_iter)
|
|
assert len(r_list) == 1
|
|
r = r_list[0]
|
|
return f"Read resource: {r.content} with mime type {r.mime_type}"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("tool_with_resource", {})
|
|
assert len(result) == 1
|
|
content = result[0]
|
|
assert isinstance(content, TextContent)
|
|
assert "Read resource: resource data" in content.text
|
|
|
|
async def test_tool_decorator_with_tags(self):
|
|
"""Test that the tool decorator properly sets tags."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(tags={"example", "test-tag"})
|
|
def sample_tool(x: int) -> int:
|
|
return x * 2
|
|
|
|
# Verify the tool exists
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
assert len(tools) == 1
|
|
# Note: MCPTool from the client API doesn't expose tags
|
|
|
|
|
|
class TestPrompts:
|
|
"""Test prompt functionality in FastMCP server."""
|
|
|
|
async def test_prompt_decorator(self):
|
|
"""Test that the prompt decorator registers prompts correctly."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt()
|
|
def fn() -> str:
|
|
return "Hello, world!"
|
|
|
|
prompts_dict = await mcp.get_prompts()
|
|
assert len(prompts_dict) == 1
|
|
prompt = prompts_dict["fn"]
|
|
assert prompt.name == "fn"
|
|
# Don't compare functions directly since validate_call wraps them
|
|
content = await prompt.render()
|
|
assert isinstance(content[0].content, TextContent)
|
|
assert content[0].content.text == "Hello, world!"
|
|
|
|
async def test_prompt_decorator_with_name(self):
|
|
"""Test prompt decorator with custom name."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(name="custom_name")
|
|
def fn() -> str:
|
|
return "Hello, world!"
|
|
|
|
prompts_dict = await mcp.get_prompts()
|
|
assert len(prompts_dict) == 1
|
|
prompt = prompts_dict["custom_name"]
|
|
assert prompt.name == "custom_name"
|
|
content = await prompt.render()
|
|
assert isinstance(content[0].content, TextContent)
|
|
assert content[0].content.text == "Hello, world!"
|
|
|
|
async def test_prompt_decorator_with_description(self):
|
|
"""Test prompt decorator with custom description."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(description="A custom description")
|
|
def fn() -> str:
|
|
return "Hello, world!"
|
|
|
|
prompts_dict = await mcp.get_prompts()
|
|
assert len(prompts_dict) == 1
|
|
prompt = prompts_dict["fn"]
|
|
assert prompt.description == "A custom description"
|
|
content = await prompt.render()
|
|
assert isinstance(content[0].content, TextContent)
|
|
assert content[0].content.text == "Hello, world!"
|
|
|
|
def test_prompt_decorator_error(self):
|
|
"""Test error when decorator is used incorrectly."""
|
|
mcp = FastMCP()
|
|
with pytest.raises(TypeError, match="decorator was used incorrectly"):
|
|
|
|
@mcp.prompt # type: ignore
|
|
def fn() -> str:
|
|
return "Hello, world!"
|
|
|
|
async def test_list_prompts(self):
|
|
"""Test listing prompts through MCP protocol."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt()
|
|
def fn(name: str, optional: str = "default") -> str:
|
|
return f"Hello, {name}! {optional}"
|
|
|
|
prompts_dict = await mcp.get_prompts()
|
|
assert len(prompts_dict) == 1
|
|
|
|
async with Client(mcp) as client:
|
|
prompts = await client.list_prompts()
|
|
assert len(prompts) == 1
|
|
assert prompts[0].name == "fn"
|
|
assert prompts[0].description is None
|
|
assert prompts[0].arguments is not None
|
|
assert len(prompts[0].arguments) == 2
|
|
assert prompts[0].arguments[0].name == "name"
|
|
assert prompts[0].arguments[0].required is True
|
|
assert prompts[0].arguments[1].name == "optional"
|
|
assert prompts[0].arguments[1].required is False
|
|
|
|
async def test_get_prompt(self):
|
|
"""Test getting a prompt through MCP protocol."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt()
|
|
def fn(name: str) -> str:
|
|
return f"Hello, {name}!"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt("fn", {"name": "World"})
|
|
assert len(result) == 1
|
|
message = result[0]
|
|
assert message.role == "user"
|
|
content = message.content
|
|
assert isinstance(content, TextContent)
|
|
assert content.text == "Hello, World!"
|
|
|
|
async def test_get_prompt_with_resource(self):
|
|
"""Test getting a prompt that returns resource content."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt()
|
|
def fn() -> Message:
|
|
return UserMessage(
|
|
content=EmbeddedResource(
|
|
type="resource",
|
|
resource=TextResourceContents(
|
|
uri=AnyUrl("file://file.txt"),
|
|
text="File contents",
|
|
mimeType="text/plain",
|
|
),
|
|
)
|
|
)
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt("fn")
|
|
assert result[0].role == "user"
|
|
content = result[0].content
|
|
assert isinstance(content, EmbeddedResource)
|
|
resource = content.resource
|
|
assert isinstance(resource, TextResourceContents)
|
|
assert resource.text == "File contents"
|
|
assert resource.mimeType == "text/plain"
|
|
|
|
async def test_get_unknown_prompt(self):
|
|
"""Test error when getting unknown prompt."""
|
|
mcp = FastMCP()
|
|
with pytest.raises(ClientError, match="Unknown prompt"):
|
|
async with Client(mcp) as client:
|
|
await client.get_prompt("unknown")
|
|
|
|
async def test_get_prompt_missing_args(self):
|
|
"""Test error when required arguments are missing."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt()
|
|
def prompt_fn(name: str) -> str:
|
|
return f"Hello, {name}!"
|
|
|
|
with pytest.raises(ClientError, match="Missing required arguments"):
|
|
async with Client(mcp) as client:
|
|
await client.get_prompt("prompt_fn")
|
|
|
|
async def test_resource_decorator_with_tags(self):
|
|
"""Test that the resource decorator supports tags."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://data", tags={"example", "test-tag"})
|
|
def get_data() -> str:
|
|
return "Hello, world!"
|
|
|
|
resources_dict = await mcp.get_resources()
|
|
resources = list(resources_dict.values())
|
|
assert len(resources) == 1
|
|
assert resources[0].tags == {"example", "test-tag"}
|
|
|
|
async def test_template_decorator_with_tags(self):
|
|
"""Test that the template decorator properly sets tags."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("resource://{param}", tags={"template", "test-tag"})
|
|
def template_resource(param: str) -> str:
|
|
return f"Template resource: {param}"
|
|
|
|
templates_dict = await mcp.get_resource_templates()
|
|
template = templates_dict["resource://{param}"]
|
|
assert template.tags == {"template", "test-tag"}
|
|
|
|
async def test_prompt_decorator_with_tags(self):
|
|
"""Test that the prompt decorator properly sets tags."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(tags={"example", "test-tag"})
|
|
def sample_prompt() -> str:
|
|
return "Hello, world!"
|
|
|
|
prompts_dict = await mcp.get_prompts()
|
|
assert len(prompts_dict) == 1
|
|
prompt = prompts_dict["sample_prompt"]
|
|
assert prompt.tags == {"example", "test-tag"}
|