Replace type: ignore[attr-defined] with isinstance assertions in tests (#2665)

* Replace type: ignore[attr-defined] with isinstance assertions in tests

* Fix isinstance assertions in failing tests

- Fix enum test to check for ResponseEnum instead of str
- Fix binary resource test to check for BlobResourceContents instead of TextResourceContents
- Fix Root type tests to check attributes directly instead of isinstance checks

* Fix type errors without using type: ignore

- Remove execution methods from TransformingProvider (only handles transformations)
- Add execution methods to base Provider class with default implementations
- Fix type narrowing in tests using cast() instead of type: ignore
- Fix PromptResult type handling in prompt render tests
- Fix type narrowing in middleware test for arguments and structured_content
This commit is contained in:
Jeremiah Lowin 2025-12-21 16:37:20 -05:00 committed by GitHub
commit 049bd22827
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
43 changed files with 817 additions and 409 deletions

View file

@ -130,7 +130,6 @@ async with Client(transport=StreamableHttpTransport(server_url)) as client:
- Follow existing patterns and maintain consistency
- **Prioritize readable, understandable code** - clarity over cleverness
- Avoid obfuscated or confusing patterns even if they're shorter
- Use `# type: ignore[attr-defined]` in tests for MCP results instead of type assertions
- Each feature needs corresponding tests
### Module Exports
@ -264,7 +263,6 @@ uv sync # Installs all deps including dev tools
### Error Handling
- Never use bare `except` - be specific with exception types
- Use `# type: ignore[attr-defined]` in tests for MCP results
### Build Issues (Common Solutions)

View file

@ -31,11 +31,12 @@ from __future__ import annotations
from collections.abc import AsyncIterator, Sequence
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Any
from fastmcp.prompts.prompt import Prompt
from fastmcp.resources.resource import Resource
from fastmcp.prompts.prompt import Prompt, PromptResult
from fastmcp.resources.resource import Resource, ResourceContent
from fastmcp.resources.template import ResourceTemplate
from fastmcp.tools.tool import Tool
from fastmcp.tools.tool import Tool, ToolResult
@dataclass
@ -228,6 +229,83 @@ class Provider:
prompts = await self.list_prompts()
return next((p for p in prompts if p.name == name), None)
# -------------------------------------------------------------------------
# Execution methods (optional - default implementations delegate to components)
# -------------------------------------------------------------------------
async def call_tool(
self, name: str, arguments: dict[str, Any]
) -> ToolResult | None:
"""Call a tool by name.
Default implementation gets the tool and calls its run() method.
Override for custom execution logic.
Returns:
ToolResult if the tool was found and executed, None otherwise.
"""
tool = await self.get_tool(name)
if tool is None:
return None
return await tool.run(arguments)
async def read_resource(self, uri: str) -> ResourceContent | None:
"""Read a resource by URI.
Default implementation gets the resource and calls its read() method.
Override for custom execution logic.
Returns:
ResourceContent if the resource was found and read, None otherwise.
"""
resource = await self.get_resource(uri)
if resource is None:
return None
result = await resource.read()
if isinstance(result, ResourceContent):
return result
return ResourceContent.from_value(result)
async def read_resource_template(self, uri: str) -> ResourceContent | None:
"""Read a resource template by URI.
Default implementation gets the template, extracts parameters from the URI,
and calls its read() method with those parameters.
Override for custom execution logic.
Returns:
ResourceContent if the template was found and read, None otherwise.
"""
template = await self.get_resource_template(uri)
if template is None:
return None
params = template.matches(uri)
if params is None:
return None
result = await template.read(params)
if isinstance(result, ResourceContent):
return result
return ResourceContent.from_value(result)
async def render_prompt(
self, name: str, arguments: dict[str, Any] | None
) -> PromptResult | None:
"""Render a prompt by name.
Default implementation gets the prompt and calls its render() method.
Override for custom execution logic.
Returns:
PromptResult if the prompt was found and rendered, None otherwise.
"""
prompt = await self.get_prompt(name)
if prompt is None:
return None
result = await prompt.render(arguments)
if isinstance(result, PromptResult):
return result
return PromptResult.from_value(result)
# -------------------------------------------------------------------------
# Task registration
# -------------------------------------------------------------------------

View file

@ -9,13 +9,13 @@ from __future__ import annotations
import re
from collections.abc import AsyncIterator, Sequence
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING
from fastmcp.prompts.prompt import Prompt, PromptResult
from fastmcp.resources.resource import Resource, ResourceContent
from fastmcp.prompts.prompt import Prompt
from fastmcp.resources.resource import Resource
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.providers.base import Provider, TaskComponents
from fastmcp.tools.tool import Tool, ToolResult
from fastmcp.tools.tool import Tool
if TYPE_CHECKING:
from fastmcp.prompts.prompt import FunctionPrompt
@ -74,7 +74,7 @@ class TransformingProvider(Provider):
use the specified name instead of namespace prefixing.
"""
super().__init__()
self._wrapped = provider
self._wrapped: Provider = provider
self.namespace = namespace
self.tool_renames = tool_renames or {}
@ -186,15 +186,6 @@ class TransformingProvider(Provider):
return tool.model_copy(update={"name": name})
return None
async def call_tool(
self, name: str, arguments: dict[str, Any]
) -> ToolResult | None:
"""Call tool by transformed name."""
original = self._reverse_tool_name(name)
if original is None:
return None
return await self._wrapped.call_tool(original, arguments)
# -------------------------------------------------------------------------
# Resource methods
# -------------------------------------------------------------------------
@ -217,13 +208,6 @@ class TransformingProvider(Provider):
return resource.model_copy(update={"uri": uri})
return None
async def read_resource(self, uri: str) -> ResourceContent | None:
"""Read resource by transformed URI."""
original = self._reverse_resource_uri(uri)
if original is None:
return None
return await self._wrapped.read_resource(original)
# -------------------------------------------------------------------------
# Resource template methods
# -------------------------------------------------------------------------
@ -252,13 +236,6 @@ class TransformingProvider(Provider):
)
return None
async def read_resource_template(self, uri: str) -> ResourceContent | None:
"""Read resource template by transformed URI."""
original = self._reverse_resource_uri(uri)
if original is None:
return None
return await self._wrapped.read_resource_template(original)
# -------------------------------------------------------------------------
# Prompt methods
# -------------------------------------------------------------------------
@ -281,15 +258,6 @@ class TransformingProvider(Provider):
return prompt.model_copy(update={"name": name})
return None
async def render_prompt(
self, name: str, arguments: dict[str, Any] | None
) -> PromptResult | None:
"""Render prompt by transformed name."""
original = self._reverse_prompt_name(name)
if original is None:
return None
return await self._wrapped.render_prompt(original, arguments)
# -------------------------------------------------------------------------
# Task registration
# -------------------------------------------------------------------------

View file

@ -48,6 +48,7 @@ class TestVersionCommand:
def test_version_command_parsing(self):
"""Test that the version command parses arguments correctly."""
command, bound, _ = app.parse_args(["version"])
assert callable(command)
assert command.__name__ == "version" # type: ignore[attr-defined]
# Default arguments aren't included in bound.arguments
assert bound.arguments == {}
@ -55,6 +56,7 @@ class TestVersionCommand:
def test_version_command_with_copy_flag(self):
"""Test that the version command parses --copy flag correctly."""
command, bound, _ = app.parse_args(["version", "--copy"])
assert callable(command)
assert command.__name__ == "version" # type: ignore[attr-defined]
assert bound.arguments == {"copy": True}

View file

@ -30,6 +30,7 @@ class TestWorkerCommand:
def test_worker_command_parsing(self):
"""Test that worker command parses arguments correctly."""
command, bound, _ = tasks_app.parse_args(["worker", "server.py"])
assert callable(command)
assert command.__name__ == "worker" # type: ignore[attr-defined]
assert bound.arguments["server_spec"] == "server.py"

View file

@ -2,6 +2,7 @@ from urllib.parse import urlparse
import httpx
import pytest
from mcp.types import TextResourceContents
from fastmcp.client import Client
from fastmcp.client.auth import OAuth
@ -103,7 +104,8 @@ async def test_read_resource(client_with_headless_oauth: Client):
"""Test that we can read a resource."""
async with client_with_headless_oauth:
resource = await client_with_headless_oauth.read_resource("resource://test")
assert resource[0].text == "Hello from authenticated resource!" # type: ignore[attr-defined]
assert isinstance(resource[0], TextResourceContents)
assert resource[0].text == "Hello from authenticated resource!"
async def test_oauth_server_metadata_discovery(streamable_http_server: str):

View file

@ -4,6 +4,7 @@ Regression test for: PromptMessage object has no attribute 'to_mcp'
"""
import mcp.types
from mcp.types import TextContent
from fastmcp import FastMCP
from fastmcp.client import Client
@ -25,7 +26,8 @@ async def test_prompt_task_with_mcp_prompt_message():
async with Client(mcp_server) as client:
task = await client.get_prompt("greeting", {"name": "World"}, task=True)
result = await task.result()
assert "Hello World" in result.messages[0].content.text # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert "Hello World" in result.messages[0].content.text
async def test_prompt_task_with_multiple_mcp_prompt_messages():
@ -53,5 +55,7 @@ async def test_prompt_task_with_multiple_mcp_prompt_messages():
task = await client.get_prompt("conversation", {"topic": "space"}, task=True)
result = await task.result()
assert len(result.messages) == 2
assert "Tell me about space" in result.messages[0].content.text # type: ignore[attr-defined]
assert "space is fascinating" in result.messages[1].content.text # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert "Tell me about space" in result.messages[0].content.text
assert isinstance(result.messages[1].content, TextContent)
assert "space is fascinating" in result.messages[1].content.text

View file

@ -8,6 +8,7 @@ import anyio
import pytest
from mcp import ClientSession, McpError
from mcp.client.auth import OAuthClientProvider
from mcp.types import TextContent
from pydantic import AnyUrl
import fastmcp
@ -124,7 +125,8 @@ async def test_call_tool(fastmcp_server):
async with client:
result = await client.call_tool("greet", {"name": "World"})
assert result.content[0].text == "Hello, World!" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Hello, World!"
assert result.structured_content == {"result": "Hello, World!"}
assert result.data == "Hello, World!"
assert result.is_error is False
@ -249,7 +251,8 @@ async def test_get_prompt(fastmcp_server):
result = await client.get_prompt("welcome", {"name": "Developer"})
# The result should contain our welcome message
assert result.messages[0].content.text == "Welcome to FastMCP, Developer!" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Welcome to FastMCP, Developer!"
assert result.description == "Example greeting prompt."
@ -261,7 +264,8 @@ async def test_get_prompt_mcp(fastmcp_server):
result = await client.get_prompt_mcp("welcome", {"name": "Developer"})
# The result should contain our welcome message
assert result.messages[0].content.text == "Welcome to FastMCP, Developer!" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Welcome to FastMCP, Developer!"
assert result.description == "Example greeting prompt."
@ -286,7 +290,8 @@ async def test_client_serializes_all_non_string_arguments():
},
)
content = result.messages[0].content.text # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
content = result.messages[0].content.text
assert "arg1: hello" in content
assert "arg2: [1,2,3]" in content # JSON serialized list
assert 'arg3: {"key":"value"}' in content # JSON serialized dict
@ -309,7 +314,8 @@ async def test_client_server_type_conversion_integration():
{"numbers": [1, 2, 3, 4], "config": {"theme": "dark", "lang": "en"}},
)
content = result.messages[0].content.text # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
content = result.messages[0].content.text
assert "Got 4 numbers and 2 config items" in content
@ -800,8 +806,9 @@ class TestErrorHandling:
async with client:
result = await client.call_tool_mcp("error_tool", {})
assert result.isError
assert "test error" in result.content[0].text # type: ignore[attr-defined]
assert "abc" in result.content[0].text # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "test error" in result.content[0].text
assert "abc" in result.content[0].text
async def test_general_tool_exceptions_are_masked_when_enabled(self):
mcp = FastMCP("TestServer", mask_error_details=True)
@ -815,8 +822,9 @@ class TestErrorHandling:
async with client:
result = await client.call_tool_mcp("error_tool", {})
assert result.isError
assert "test error" not in result.content[0].text # type: ignore[attr-defined]
assert "abc" not in result.content[0].text # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "test error" not in result.content[0].text
assert "abc" not in result.content[0].text
async def test_validation_errors_are_not_masked_when_enabled(self):
mcp = FastMCP("TestServer", mask_error_details=True)
@ -829,7 +837,8 @@ class TestErrorHandling:
result = await client.call_tool_mcp("validated_tool", {"x": "abc"})
assert result.isError
# Pydantic validation error message should NOT be masked
assert "Input should be a valid integer" in result.content[0].text # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "Input should be a valid integer" in result.content[0].text
async def test_specific_tool_errors_are_sent_to_client(self):
mcp = FastMCP("TestServer")
@ -843,8 +852,9 @@ class TestErrorHandling:
async with client:
result = await client.call_tool_mcp("custom_error_tool", {})
assert result.isError
assert "test error" in result.content[0].text # type: ignore[attr-defined]
assert "abc" in result.content[0].text # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "test error" in result.content[0].text
assert "abc" in result.content[0].text
async def test_general_resource_exceptions_are_not_masked_by_default(self):
mcp = FastMCP("TestServer")

View file

@ -1,6 +1,6 @@
from dataclasses import asdict, dataclass
from enum import Enum
from typing import Literal
from typing import Any, Literal, cast
import pytest
from mcp.types import ElicitRequestFormParams, ElicitRequestParams
@ -36,7 +36,9 @@ def fastmcp_server():
response_type=Person,
)
if result.action == "accept":
return f"Hello, {result.data.name}!" # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, Person)
return f"Hello, {result.data.name}!"
else:
return "No name provided."
@ -128,7 +130,9 @@ async def test_elicitation_cancel_action():
if result.action == "cancel":
return "Request was canceled"
elif result.action == "accept":
return f"Age: {result.data}" # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, int)
return f"Age: {result.data}"
else:
return "No response provided"
@ -146,9 +150,11 @@ class TestScalarResponseTypes:
mcp = FastMCP("TestServer")
@mcp.tool
async def my_tool(context: Context) -> None:
async def my_tool(context: Context) -> dict[str, Any]:
result = await context.elicit(message="", response_type=None)
return result.data # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, dict)
return cast(dict[str, Any], result.data)
async def elicitation_handler(
message, response_type, params: ElicitRequestParams, ctx
@ -167,9 +173,13 @@ class TestScalarResponseTypes:
mcp = FastMCP("TestServer")
@mcp.tool
async def my_tool(context: Context) -> None:
async def my_tool(context: Context) -> dict[str, Any]:
result = await context.elicit(message="", response_type=None)
return result.data # type: ignore[attr-defined]
assert result.action == "accept"
assert isinstance(result, AcceptedElicitation)
accepted = cast(AcceptedElicitation[dict[str, Any]], result)
assert isinstance(accepted.data, dict)
return accepted.data
async def elicitation_handler(
message, response_type, params: ElicitRequestParams, ctx
@ -185,9 +195,13 @@ class TestScalarResponseTypes:
mcp = FastMCP("TestServer")
@mcp.tool
async def my_tool(context: Context) -> None:
async def my_tool(context: Context) -> dict[str, Any]:
result = await context.elicit(message="", response_type=None)
return result.data # type: ignore[attr-defined]
assert result.action == "accept"
assert isinstance(result, AcceptedElicitation)
accepted = cast(AcceptedElicitation[dict[str, Any]], result)
assert isinstance(accepted.data, dict)
return accepted.data
async def elicitation_handler(message, response_type, params, ctx):
return ElicitResult(action="accept", content={"value": "hello"})
@ -205,7 +219,9 @@ class TestScalarResponseTypes:
@mcp.tool
async def my_tool(context: Context) -> str:
result = await context.elicit(message="", response_type=str)
return result.data # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, str)
return result.data
async def elicitation_handler(message, response_type, params, ctx):
return ElicitResult(action="accept", content={"value": "hello"})
@ -221,7 +237,9 @@ class TestScalarResponseTypes:
@mcp.tool
async def my_tool(context: Context) -> int:
result = await context.elicit(message="", response_type=int)
return result.data # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, int)
return result.data
async def elicitation_handler(message, response_type, params, ctx):
return ElicitResult(action="accept", content={"value": 42})
@ -237,7 +255,9 @@ class TestScalarResponseTypes:
@mcp.tool
async def my_tool(context: Context) -> float:
result = await context.elicit(message="", response_type=float)
return result.data # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, float)
return result.data
async def elicitation_handler(message, response_type, params, ctx):
return ElicitResult(action="accept", content={"value": 3.14})
@ -253,7 +273,9 @@ class TestScalarResponseTypes:
@mcp.tool
async def my_tool(context: Context) -> bool:
result = await context.elicit(message="", response_type=bool)
return result.data # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, bool)
return result.data
async def elicitation_handler(message, response_type, params, ctx):
return ElicitResult(action="accept", content={"value": True})
@ -268,8 +290,12 @@ class TestScalarResponseTypes:
@mcp.tool
async def my_tool(context: Context) -> Literal["x", "y"]:
result = await context.elicit(message="", response_type=Literal["x", "y"]) # type: ignore
return result.data # type: ignore[attr-defined]
# Literal types work at runtime but type checker doesn't recognize them in overloads
result = await context.elicit(message="", response_type=Literal["x", "y"]) # type: ignore[arg-type]
assert isinstance(result, AcceptedElicitation)
accepted = cast(AcceptedElicitation[Literal["x", "y"]], result)
assert isinstance(accepted.data, str)
return accepted.data
async def elicitation_handler(message, response_type, params, ctx):
return ElicitResult(action="accept", content={"value": "x"})
@ -289,7 +315,9 @@ class TestScalarResponseTypes:
@mcp.tool
async def my_tool(context: Context) -> ResponseEnum:
result = await context.elicit(message="", response_type=ResponseEnum)
return result.data # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, ResponseEnum)
return result.data
async def elicitation_handler(message, response_type, params, ctx):
return ElicitResult(action="accept", content={"value": "x"})
@ -305,7 +333,9 @@ class TestScalarResponseTypes:
@mcp.tool
async def my_tool(context: Context) -> str:
result = await context.elicit(message="", response_type=["x", "y"])
return result.data # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, str)
return result.data
async def elicitation_handler(message, response_type, params, ctx):
return ElicitResult(action="accept", content={"value": "x"})
@ -417,10 +447,19 @@ async def test_structured_response_type(
assert isinstance(result, AcceptedElicitation)
if result.action == "accept":
assert isinstance(result, AcceptedElicitation)
if isinstance(result.data, dict):
return f"User: {result.data['name']}, age: {result.data['age']}" # type: ignore[index]
data_dict = cast(dict[str, Any], result.data)
name = data_dict.get("name")
age = data_dict.get("age")
assert name is not None
assert age is not None
return f"User: {name}, age: {age}"
else:
return f"User: {result.data.name}, age: {result.data.age}" # type: ignore[attr-defined]
# result.data is a structured type (UserInfo, UserInfoTypedDict, or UserInfoPydantic)
assert hasattr(result.data, "name")
assert hasattr(result.data, "age")
return f"User: {result.data.name}, age: {result.data.age}"
return "No user info provided"
async def elicitation_handler(message, response_type, params, ctx):
@ -467,7 +506,9 @@ async def test_all_primitive_field_types():
@mcp.tool
async def get_data(context: Context) -> Data:
result = await context.elicit(message="Enter data", response_type=Data)
return result.data # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, Data)
return result.data
async def elicitation_handler(message, response_type, params, ctx):
return ElicitResult(
@ -738,7 +779,9 @@ async def test_dict_based_titled_single_select():
},
)
if result.action == "accept":
return result.data # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, str)
return result.data
return "declined"
async def elicitation_handler(message, response_type, params, ctx):
@ -760,7 +803,9 @@ async def test_list_list_multi_select_untitled():
response_type=[["bug", "feature", "documentation"]],
)
if result.action == "accept":
return ",".join(result.data) # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, list)
return ",".join(result.data)
return "declined"
async def elicitation_handler(message, response_type, params, ctx):
@ -796,7 +841,9 @@ async def test_list_dict_multi_select_titled():
],
)
if result.action == "accept":
return ",".join(result.data) # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, list)
return ",".join(result.data)
return "declined"
async def elicitation_handler(message, response_type, params, ctx):
@ -857,7 +904,9 @@ async def test_list_enum_multi_select_direct():
response_type=list[Priority], # Type annotation for multi-select
)
if result.action == "accept":
priorities = result.data # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, list)
priorities = result.data
return ",".join(
[p.value if isinstance(p, Priority) else str(p) for p in priorities]
)

View file

@ -2,6 +2,7 @@ import json
import pytest
from fastapi import FastAPI, Request
from mcp.types import TextResourceContents
from fastmcp import Client, FastMCP
from fastmcp.client.transports import SSETransport, StreamableHttpTransport
@ -70,14 +71,16 @@ async def proxy_server(shttp_server: str):
async def test_fastapi_client_headers_streamable_http_resource(shttp_server: str):
async with Client(transport=StreamableHttpTransport(shttp_server)) as client:
result = await client.read_resource("resource://get_headers_headers_get")
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
assert headers["x-server-header"] == "test-abc"
async def test_fastapi_client_headers_sse_resource(sse_server: str):
async with Client(transport=SSETransport(sse_server)) as client:
result = await client.read_resource("resource://get_headers_headers_get")
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
assert headers["x-server-header"] == "test-abc"
@ -100,7 +103,8 @@ async def test_client_headers_sse_resource(sse_server: str):
transport=SSETransport(sse_server, headers={"X-TEST": "test-123"})
) as client:
result = await client.read_resource("resource://get_headers_headers_get")
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
assert headers["x-test"] == "test-123"
@ -109,7 +113,8 @@ async def test_client_headers_shttp_resource(shttp_server: str):
transport=StreamableHttpTransport(shttp_server, headers={"X-TEST": "test-123"})
) as client:
result = await client.read_resource("resource://get_headers_headers_get")
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
assert headers["x-test"] == "test-123"
@ -120,7 +125,8 @@ async def test_client_headers_sse_resource_template(sse_server: str):
result = await client.read_resource(
"resource://get_header_by_name_headers/x-test"
)
header = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
header = json.loads(result[0].text)
assert header == "test-123"
@ -131,7 +137,8 @@ async def test_client_headers_shttp_resource_template(shttp_server: str):
result = await client.read_resource(
"resource://get_header_by_name_headers/x-test"
)
header = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
header = json.loads(result[0].text)
assert header == "test-123"
@ -160,7 +167,8 @@ async def test_client_overrides_server_headers(shttp_server: str):
)
) as client:
result = await client.read_resource("resource://get_headers_headers_get")
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
assert headers["x-server-header"] == "test-client"
@ -176,7 +184,8 @@ async def test_client_with_excluded_header_is_ignored(sse_server: str):
)
) as client:
result = await client.read_resource("resource://get_headers_headers_get")
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
assert headers["not-host"] == "1.2.3.4"
assert headers["host"] == "fastapi"
@ -188,5 +197,6 @@ async def test_client_headers_proxy(proxy_server: str):
"""
async with Client(transport=StreamableHttpTransport(proxy_server)) as client:
result = await client.read_resource("resource://get_headers_headers_get")
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
assert headers["x-server-header"] == "test-abc"

View file

@ -8,7 +8,7 @@ from pydantic_core import to_json
from fastmcp import Client, Context, FastMCP
from fastmcp.client.sampling import RequestContext, SamplingMessage, SamplingParams
from fastmcp.server.sampling import SamplingTool
from fastmcp.server.sampling import SamplingResult, SamplingTool
from fastmcp.utilities.types import Image
@ -19,12 +19,16 @@ def fastmcp_server():
@mcp.tool
async def simple_sample(message: str, context: Context) -> str:
result = await context.sample("Hello, world!")
return result.text # type: ignore[attr-defined]
assert isinstance(result, SamplingResult)
assert result.text is not None
return result.text
@mcp.tool
async def sample_with_system_prompt(message: str, context: Context) -> str:
result = await context.sample("Hello, world!", system_prompt="You love FastMCP")
return result.text # type: ignore[attr-defined]
assert isinstance(result, SamplingResult)
assert result.text is not None
return result.text
@mcp.tool
async def sample_with_messages(message: str, context: Context) -> str:
@ -39,7 +43,9 @@ def fastmcp_server():
),
]
)
return result.text # type: ignore[attr-defined]
assert isinstance(result, SamplingResult)
assert result.text is not None
return result.text
@mcp.tool
async def sample_with_image(image_bytes: bytes, context: Context) -> str:
@ -57,7 +63,9 @@ def fastmcp_server():
),
]
)
return result.text # type: ignore[attr-defined]
assert isinstance(result, SamplingResult)
assert result.text is not None
return result.text
return mcp
@ -476,7 +484,8 @@ class TestAutomaticToolLoop:
assert tool_result is not None
assert tool_result.isError is True
# Content is list of TextContent objects
error_text = tool_result.content[0].text # type: ignore[union-attr]
assert isinstance(tool_result.content[0], TextContent)
error_text = tool_result.content[0].text
assert "Unknown tool" in error_text
assert result.data == "Handled error"
@ -549,7 +558,8 @@ class TestAutomaticToolLoop:
assert tool_result is not None
assert tool_result.isError is True
# Content is list of TextContent objects
error_text = tool_result.content[0].text # type: ignore[union-attr]
assert isinstance(tool_result.content[0], TextContent)
error_text = tool_result.content[0].text
assert "Tool failed intentionally" in error_text
assert result.data == "Handled error"
@ -597,7 +607,8 @@ class TestSamplingResultType:
result_type=MathResult,
)
# result.result should be a MathResult object
return f"{result.result.answer}: {result.result.explanation}" # type: ignore[attr-defined]
assert isinstance(result.result, MathResult)
return f"{result.result.answer}: {result.result.explanation}"
async with Client(mcp) as client:
result = await client.call_tool("math_tool", {})
@ -675,7 +686,8 @@ class TestSamplingResultType:
tools=[search],
result_type=SearchResult,
)
return f"{result.result.summary} - {len(result.result.sources)} sources" # type: ignore[attr-defined]
assert isinstance(result.result, SearchResult)
return f"{result.result.summary} - {len(result.result.sources)} sources"
async with Client(mcp) as client:
result = await client.call_tool("research", {})
@ -741,7 +753,8 @@ class TestSamplingResultType:
messages="Give me a number",
result_type=StrictResult,
)
return str(result.result.value) # type: ignore[attr-defined]
assert isinstance(result.result, StrictResult)
return str(result.result.value)
async with Client(mcp) as client:
result = await client.call_tool("validate_tool", {})
@ -765,7 +778,8 @@ class TestSamplingResultType:
break
assert tool_result is not None
assert tool_result.isError is True
error_text = tool_result.content[0].text # type: ignore[union-attr]
assert isinstance(tool_result.content[0], TextContent)
error_text = tool_result.content[0].text
assert "Validation error" in error_text
# Final result should be correct

View file

@ -4,6 +4,7 @@ import sys
import pytest
from mcp import McpError
from mcp.types import TextResourceContents
from fastmcp.client import Client
from fastmcp.client.transports import SSETransport
@ -75,7 +76,8 @@ async def test_http_headers(sse_server: str):
transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
) as client:
raw_result = await client.read_resource("request://headers")
json_result = json.loads(raw_result[0].text) # type: ignore[attr-defined]
assert isinstance(raw_result[0], TextResourceContents)
json_result = json.loads(raw_result[0].text)
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"

View file

@ -5,6 +5,7 @@ from unittest.mock import AsyncMock, call
import pytest
from mcp import McpError
from mcp.types import TextResourceContents
from fastmcp import Context
from fastmcp.client import Client
@ -170,7 +171,8 @@ async def test_http_headers(streamable_http_server: str):
)
) as client:
raw_result = await client.read_resource("request://headers")
json_result = json.loads(raw_result[0].text) # type: ignore[attr-defined]
assert isinstance(raw_result[0], TextResourceContents)
json_result = json.loads(raw_result[0].text)
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"

View file

@ -2,6 +2,7 @@ from ssl import VerifyMode
import httpx
from fastmcp.client.auth.oauth import OAuth
from fastmcp.client.transports import SSETransport, StreamableHttpTransport
@ -14,7 +15,8 @@ async def test_oauth_uses_same_client_as_transport_streamable_http():
auth="oauth",
)
async with transport.auth.httpx_client_factory() as httpx_client: # type: ignore[attr-defined]
assert isinstance(transport.auth, OAuth)
async with transport.auth.httpx_client_factory() as httpx_client:
assert httpx_client._transport is not None
assert (
httpx_client._transport._pool._ssl_context.verify_mode # type: ignore[attr-defined]
@ -31,7 +33,8 @@ async def test_oauth_uses_same_client_as_transport_sse():
auth="oauth",
)
async with transport.auth.httpx_client_factory() as httpx_client: # type: ignore[attr-defined]
assert isinstance(transport.auth, OAuth)
async with transport.auth.httpx_client_factory() as httpx_client:
assert httpx_client._transport is not None
assert (
httpx_client._transport._pool._ssl_context.verify_mode # type: ignore[attr-defined]

View file

@ -1,6 +1,8 @@
import json
from urllib.parse import quote
from mcp.types import TextContent, TextResourceContents
from fastmcp.client.client import Client
from fastmcp.server.server import FastMCP
from fastmcp.tools.tool import FunctionTool, Tool
@ -328,7 +330,8 @@ async def test_import_with_proxy_prompts():
async with Client(main_app) as client:
result = await client.get_prompt("api_greeting", {"name": "World"})
assert result.messages[0].content.text == "Hello, World from API!" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Hello, World from API!"
assert result.description == "Example greeting prompt."
@ -357,7 +360,8 @@ async def test_import_with_proxy_resources():
# Access the resource through the main app with the prefixed key
async with Client(main_app) as client:
result = await client.read_resource("config://api/settings")
content = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
content = json.loads(result[0].text)
assert content["api_key"] == "12345"
assert content["base_url"] == "https://api.example.com"
@ -387,7 +391,8 @@ async def test_import_with_proxy_resource_templates():
quoted_email = quote("john@example.com", safe="")
async with Client(main_app) as client:
result = await client.read_resource(f"user://api/{quoted_name}/{quoted_email}")
content = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
content = json.loads(result[0].text)
assert content["name"] == "John Doe"
assert content["email"] == "john@example.com"
@ -430,16 +435,19 @@ async def test_import_with_no_prefix():
# Test resource
resource_result = await client.read_resource("data://config")
assert resource_result[0].text == "Sub resource data" # type: ignore[attr-defined]
assert isinstance(resource_result[0], TextResourceContents)
assert resource_result[0].text == "Sub resource data"
# Test template
template_result = await client.read_resource("users://123/info")
assert template_result[0].text == "Sub template for user 123" # type: ignore[attr-defined]
assert isinstance(template_result[0], TextResourceContents)
assert template_result[0].text == "Sub template for user 123"
# Test prompt
prompt_result = await client.get_prompt("sub_prompt", {})
assert prompt_result.messages is not None
assert prompt_result.messages[0].content.text == "Sub prompt content" # type: ignore[attr-defined]
assert isinstance(prompt_result.messages[0].content, TextContent)
assert prompt_result.messages[0].content.text == "Sub prompt content"
async def test_import_conflict_resolution_tools():
@ -497,7 +505,8 @@ async def test_import_conflict_resolution_resources():
assert resource_uris.count("shared://data") == 1 # Should only appear once
result = await client.read_resource("shared://data")
assert result[0].text == "Second app data" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Second app data"
async def test_import_conflict_resolution_templates():
@ -528,7 +537,8 @@ async def test_import_conflict_resolution_templates():
) # Should only appear once
result = await client.read_resource("users://123/profile")
assert result[0].text == "Second app user 123" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Second app user 123"
async def test_import_conflict_resolution_prompts():
@ -558,7 +568,8 @@ async def test_import_conflict_resolution_prompts():
result = await client.get_prompt("shared_prompt", {})
assert result.messages is not None
assert result.messages[0].content.text == "Second app prompt" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Second app prompt"
async def test_import_conflict_resolution_with_prefix():
@ -660,10 +671,13 @@ async def test_import_server_with_new_prefix_format():
# Verify we can access the resources
async with Client(target_server) as client:
result = await client.read_resource("resource://imported/test-resource")
assert result[0].text == "Resource content" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Resource content"
result = await client.read_resource("resource://imported//absolute/path")
assert result[0].text == "Absolute resource content" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Absolute resource content"
result = await client.read_resource("resource://imported/param-value/template")
assert result[0].text == "Template resource with param-value" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource with param-value"

View file

@ -3,7 +3,7 @@ import os
import pytest
from mcp import McpError
from mcp.types import Tool
from mcp.types import TextContent, Tool
from fastmcp import Client
from fastmcp.client import StreamableHttpTransport
@ -111,7 +111,8 @@ class TestGithubMCPRemote:
assert result.structured_content is None
assert isinstance(result.content, list)
assert len(result.content) == 1
commits = json.loads(result.content[0].text) # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
commits = json.loads(result.content[0].text)
for commit in commits:
assert isinstance(commit, dict)
assert "sha" in commit

View file

@ -413,7 +413,8 @@ class TestContextHandling:
assert isinstance(result, PromptResult)
assert len(result.messages) == 1
assert result.messages[0].content.text == "42" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "42"
async def test_context_optional(self):
"""Test that context is optional when rendering prompts."""
@ -436,7 +437,8 @@ class TestContextHandling:
assert isinstance(result, PromptResult)
assert len(result.messages) == 1
assert result.messages[0].content.text == "42" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "42"
async def test_annotated_context_parameter_detection(self):
"""Test that annotated context parameters are properly detected in
@ -473,4 +475,5 @@ class TestContextHandling:
async with context:
result = await prompt.render(arguments={"topic": "cats"})
assert isinstance(result, PromptResult)
assert result.messages[0].content.text == "Write about cats" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Write about cats"

View file

@ -10,6 +10,7 @@ from mcp.shared.auth import OAuthClientInformationFull
from pydantic import AnyUrl
from fastmcp.server.auth.providers.azure import OIDC_SCOPES, AzureProvider
from fastmcp.server.auth.providers.jwt import JWTVerifier
class TestAzureProvider:
@ -404,12 +405,13 @@ class TestAzureProvider:
provider._upstream_token_endpoint
== "https://login.microsoftonline.com/test-tenant/oauth2/v2.0/token"
)
assert isinstance(provider._token_validator, JWTVerifier)
assert (
provider._token_validator.issuer # type: ignore[attr-defined]
provider._token_validator.issuer
== "https://login.microsoftonline.com/test-tenant/v2.0"
)
assert (
provider._token_validator.jwks_uri # type: ignore[attr-defined]
provider._token_validator.jwks_uri
== "https://login.microsoftonline.com/test-tenant/discovery/v2.0/keys"
)
@ -432,12 +434,13 @@ class TestAzureProvider:
provider._upstream_token_endpoint
== "https://login.microsoftonline.us/gov-tenant-id/oauth2/v2.0/token"
)
assert isinstance(provider._token_validator, JWTVerifier)
assert (
provider._token_validator.issuer # type: ignore[attr-defined]
provider._token_validator.issuer
== "https://login.microsoftonline.us/gov-tenant-id/v2.0"
)
assert (
provider._token_validator.jwks_uri # type: ignore[attr-defined]
provider._token_validator.jwks_uri
== "https://login.microsoftonline.us/gov-tenant-id/discovery/v2.0/keys"
)
@ -464,12 +467,13 @@ class TestAzureProvider:
provider._upstream_token_endpoint
== "https://login.microsoftonline.us/env-tenant-id/oauth2/v2.0/token"
)
assert isinstance(provider._token_validator, JWTVerifier)
assert (
provider._token_validator.issuer # type: ignore[attr-defined]
provider._token_validator.issuer
== "https://login.microsoftonline.us/env-tenant-id/v2.0"
)
assert (
provider._token_validator.jwks_uri # type: ignore[attr-defined]
provider._token_validator.jwks_uri
== "https://login.microsoftonline.us/env-tenant-id/discovery/v2.0/keys"
)

View file

@ -9,6 +9,7 @@ import pytest
from fastmcp import Client, FastMCP
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp.server.auth.providers.descope import DescopeProvider
from fastmcp.server.auth.providers.jwt import JWTVerifier
from fastmcp.utilities.tests import HeadlessOAuth, run_server_async
@ -56,8 +57,9 @@ class TestDescopeProvider:
assert provider.project_id == "P2oldenv123"
assert str(provider.base_url) == "https://envserver.com/"
assert str(provider.descope_base_url) == "https://api.descope.com"
assert isinstance(provider.token_verifier, JWTVerifier)
assert (
provider.token_verifier.issuer # type: ignore[attr-defined]
provider.token_verifier.issuer
== "https://api.descope.com/v1/apps/P2oldenv123"
)
@ -121,12 +123,12 @@ class TestDescopeProvider:
assert str(provider.base_url) == "https://myserver.com/"
# Check that JWT verifier uses the old issuer format
assert isinstance(provider.token_verifier, JWTVerifier)
assert (
provider.token_verifier.issuer # type: ignore[attr-defined]
== "https://api.descope.com/v1/apps/P2abc123"
provider.token_verifier.issuer == "https://api.descope.com/v1/apps/P2abc123"
)
assert (
provider.token_verifier.jwks_uri # type: ignore[attr-defined]
provider.token_verifier.jwks_uri
== "https://api.descope.com/P2abc123/.well-known/jwks.json"
)
@ -139,9 +141,9 @@ class TestDescopeProvider:
)
assert str(provider.descope_base_url) == "https://api.descope.com"
assert isinstance(provider.token_verifier, JWTVerifier)
assert (
provider.token_verifier.issuer # type: ignore[attr-defined]
== "https://api.descope.com/v1/apps/P2abc123"
provider.token_verifier.issuer == "https://api.descope.com/v1/apps/P2abc123"
)
def test_config_url_takes_precedence_over_old_api(self):
@ -156,8 +158,9 @@ class TestDescopeProvider:
# Should use values from config_url, not the old API
assert provider.project_id == "P2new123"
assert str(provider.descope_base_url) == "https://api.descope.com"
assert isinstance(provider.token_verifier, JWTVerifier)
assert (
provider.token_verifier.issuer # type: ignore[attr-defined]
provider.token_verifier.issuer
== "https://api.descope.com/v1/apps/agentic/P2new123/M123"
)
@ -172,14 +175,14 @@ class TestDescopeProvider:
)
# Check that JWT verifier uses the correct endpoints
assert isinstance(provider.token_verifier, JWTVerifier)
assert (
provider.token_verifier.jwks_uri # type: ignore[attr-defined]
provider.token_verifier.jwks_uri
== "https://api.descope.com/P2abc123/.well-known/jwks.json"
)
assert (
provider.token_verifier.issuer == issuer_url # type: ignore[attr-defined]
)
assert provider.token_verifier.audience == "P2abc123" # type: ignore[attr-defined]
assert provider.token_verifier.issuer == issuer_url
assert isinstance(provider.token_verifier, JWTVerifier)
assert provider.token_verifier.audience == "P2abc123"
def test_required_scopes_support(self):
"""Test that required_scopes are supported and passed to JWT verifier."""
@ -190,7 +193,8 @@ class TestDescopeProvider:
)
# Check that required_scopes are set on the token verifier
assert provider.token_verifier.required_scopes == ["read", "write"] # type: ignore[attr-defined]
assert isinstance(provider.token_verifier, JWTVerifier)
assert provider.token_verifier.required_scopes == ["read", "write"]
def test_required_scopes_with_old_api(self):
"""Test that required_scopes work with the old API (project_id + descope_base_url)."""
@ -202,7 +206,8 @@ class TestDescopeProvider:
)
# Check that required_scopes are set on the token verifier
assert provider.token_verifier.required_scopes == ["openid", "email"] # type: ignore[attr-defined]
assert isinstance(provider.token_verifier, JWTVerifier)
assert provider.token_verifier.required_scopes == ["openid", "email"]
def test_required_scopes_from_env(self):
"""Test that required_scopes can be set via environment variable."""
@ -216,7 +221,8 @@ class TestDescopeProvider:
):
provider = DescopeProvider()
assert provider.token_verifier.required_scopes == ["read", "write"] # type: ignore[attr-defined]
assert isinstance(provider.token_verifier, JWTVerifier)
assert provider.token_verifier.required_scopes == ["read", "write"]
@pytest.fixture

View file

@ -8,6 +8,7 @@ import pytest
from fastmcp import Client, FastMCP
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp.server.auth.providers.jwt import JWTVerifier
from fastmcp.server.auth.providers.scalekit import ScalekitProvider
from fastmcp.utilities.tests import HeadlessOAuth, run_server_async
@ -125,16 +126,10 @@ class TestScalekitProvider:
)
# Check that JWT verifier uses the correct endpoints
assert (
provider.token_verifier.jwks_uri # type: ignore[attr-defined]
== "https://my-env.scalekit.com/keys"
)
assert (
provider.token_verifier.issuer == "https://my-env.scalekit.com" # type: ignore[attr-defined]
)
assert (
provider.token_verifier.audience is None # type: ignore[attr-defined]
)
assert isinstance(provider.token_verifier, JWTVerifier)
assert provider.token_verifier.jwks_uri == "https://my-env.scalekit.com/keys"
assert provider.token_verifier.issuer == "https://my-env.scalekit.com"
assert provider.token_verifier.audience is None
def test_required_scopes_hooks_into_verifier(self):
"""Token verifier should enforce required scopes when provided."""
@ -145,7 +140,8 @@ class TestScalekitProvider:
required_scopes=["read"],
)
assert provider.token_verifier.required_scopes == ["read"] # type: ignore[attr-defined]
assert isinstance(provider.token_verifier, JWTVerifier)
assert provider.token_verifier.required_scopes == ["read"]
def test_authorization_servers_configuration(self):
"""Test that authorization servers are configured correctly."""

View file

@ -9,6 +9,7 @@ import pytest
from fastmcp import Client, FastMCP
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp.server.auth.providers.jwt import JWTVerifier
from fastmcp.server.auth.providers.supabase import SupabaseProvider
from fastmcp.utilities.tests import HeadlessOAuth, run_server_in_process
@ -81,14 +82,13 @@ class TestSupabaseProvider:
)
# Check that JWT verifier uses the correct endpoints
assert isinstance(provider.token_verifier, JWTVerifier)
assert (
provider.token_verifier.jwks_uri # type: ignore[attr-defined]
provider.token_verifier.jwks_uri
== "https://abc123.supabase.co/auth/v1/.well-known/jwks.json"
)
assert (
provider.token_verifier.issuer == "https://abc123.supabase.co/auth/v1" # type: ignore[attr-defined]
)
assert provider.token_verifier.algorithm == "ES256" # type: ignore[attr-defined]
assert provider.token_verifier.issuer == "https://abc123.supabase.co/auth/v1"
assert provider.token_verifier.algorithm == "ES256"
def test_jwt_verifier_with_required_scopes(self):
"""Test that JWT verifier respects required_scopes."""
@ -98,7 +98,8 @@ class TestSupabaseProvider:
required_scopes=["openid", "email"],
)
assert provider.token_verifier.required_scopes == ["openid", "email"] # type: ignore[attr-defined]
assert isinstance(provider.token_verifier, JWTVerifier)
assert provider.token_verifier.required_scopes == ["openid", "email"]
def test_authorization_servers_configured(self):
"""Test that authorization servers list is configured correctly."""
@ -125,7 +126,8 @@ class TestSupabaseProvider:
algorithm=algorithm,
)
assert provider.token_verifier.algorithm == algorithm # type: ignore[attr-defined]
assert isinstance(provider.token_verifier, JWTVerifier)
assert provider.token_verifier.algorithm == algorithm
def test_algorithm_default_es256(self):
"""Test that algorithm defaults to ES256 when not specified."""
@ -134,7 +136,8 @@ class TestSupabaseProvider:
base_url="https://myserver.com",
)
assert provider.token_verifier.algorithm == "ES256" # type: ignore[attr-defined]
assert isinstance(provider.token_verifier, JWTVerifier)
assert provider.token_verifier.algorithm == "ES256"
def test_algorithm_from_env_var(self):
"""Test that algorithm can be configured via environment variable."""
@ -148,7 +151,8 @@ class TestSupabaseProvider:
):
provider = SupabaseProvider()
assert provider.token_verifier.algorithm == "RS256" # type: ignore[attr-defined]
assert isinstance(provider.token_verifier, JWTVerifier)
assert provider.token_verifier.algorithm == "RS256"
def run_mcp_server(host: str, port: int) -> None:

View file

@ -4,7 +4,7 @@ import pytest
from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend
from starlette.requests import HTTPConnection
from fastmcp.server.auth import AccessToken
from fastmcp.server.auth import AccessToken, TokenVerifier
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
@ -41,7 +41,8 @@ class TestBearerAuthBackendTokenVerifierIntegration:
"""Test that BearerAuthBackend constructor accepts TokenVerifier."""
# This should not raise an error
backend = BearerAuthBackend(jwt_verifier)
assert backend.token_verifier is jwt_verifier # type: ignore[attr-defined]
assert isinstance(backend.token_verifier, TokenVerifier)
assert backend.token_verifier is jwt_verifier
async def test_bearer_auth_backend_authenticate_with_valid_token(
self, jwt_verifier: JWTVerifier, valid_token: str

View file

@ -1,6 +1,7 @@
import json
import pytest
from mcp.types import TextContent, TextResourceContents
from fastmcp.client import Client
from fastmcp.client.transports import SSETransport, StreamableHttpTransport
@ -61,7 +62,8 @@ async def test_http_headers_resource_shttp(shttp_server: str):
)
) as client:
raw_result = await client.read_resource("request://headers")
json_result = json.loads(raw_result[0].text) # type: ignore[attr-defined]
assert isinstance(raw_result[0], TextResourceContents)
json_result = json.loads(raw_result[0].text)
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"
@ -72,7 +74,8 @@ async def test_http_headers_resource_sse(sse_server: str):
transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
) as client:
raw_result = await client.read_resource("request://headers")
json_result = json.loads(raw_result[0].text) # type: ignore[attr-defined]
assert isinstance(raw_result[0], TextResourceContents)
json_result = json.loads(raw_result[0].text)
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"
@ -106,7 +109,8 @@ async def test_http_headers_prompt_shttp(shttp_server: str):
)
) as client:
result = await client.get_prompt("get_headers_prompt")
json_result = json.loads(result.messages[0].content.text) # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
json_result = json.loads(result.messages[0].content.text)
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"
@ -117,6 +121,7 @@ async def test_http_headers_prompt_sse(sse_server: str):
transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
) as client:
result = await client.get_prompt("get_headers_prompt")
json_result = json.loads(result.messages[0].content.text) # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
json_result = json.loads(result.messages[0].content.text)
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"

View file

@ -5,7 +5,7 @@ from typing import Any
import mcp.types as mt
import pytest
from mcp import McpError
from mcp.types import ErrorData
from mcp.types import ErrorData, TextContent
from fastmcp import Client, FastMCP
from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext
@ -138,7 +138,8 @@ async def test_middleware_receives_initialization():
# Test that the tool still works
result = await client.call_tool("test_tool", {"x": 42})
assert result.content[0].text == "Result: 42" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Result: 42"
async def test_client_detection_middleware():
@ -245,7 +246,8 @@ async def test_initialization_middleware_with_state_sharing():
# Call a tool - state should be accessible
result = await client.call_tool("test_tool", {})
assert result.content[0].text == "success" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "success"
# State should have been accessible during tool call
# Note: State is request-scoped, so it won't persist across requests

View file

@ -439,13 +439,19 @@ class TestMiddlewareHooks:
):
# modify argument
if context.message.name == "add":
context.message.arguments["a"] += 100 # type: ignore
assert context.message.arguments is not None
args = context.message.arguments
assert isinstance(args["a"], int)
args["a"] += 100
result = await call_next(context)
# modify result
if context.message.name == "add":
result.structured_content["result"] += 5 # type: ignore
assert result.structured_content is not None
content = result.structured_content
assert isinstance(content["result"], int)
content["result"] += 5
return result
@ -454,7 +460,8 @@ class TestMiddlewareHooks:
async with Client(server) as client:
result = await client.call_tool("add", {"a": 1, "b": 2})
assert result.structured_content["result"] == 108 # type: ignore
assert isinstance(result.structured_content["result"], int)
assert result.structured_content["result"] == 108
class TestNestedMiddlewareHooks:

View file

@ -93,7 +93,8 @@ class TestToolInjectionMiddleware:
)
assert result.structured_content is not None
assert result.structured_content["result"] == 42 # type: ignore[attr-defined]
assert isinstance(result.structured_content, dict)
assert result.structured_content["result"] == 42
async def test_call_base_tool_still_works(self, base_server: FastMCP):
"""Test that base server tools still work after injecting tools."""
@ -110,7 +111,8 @@ class TestToolInjectionMiddleware:
)
assert result.structured_content is not None
assert result.structured_content["result"] == 15 # type: ignore[attr-defined]
assert isinstance(result.structured_content, dict)
assert result.structured_content["result"] == 15
async def test_injected_tool_error_handling(self, base_server: FastMCP):
"""Test that errors in injected tools are properly handled."""
@ -161,11 +163,13 @@ class TestToolInjectionMiddleware:
async with Client(base_server) as client:
power_result = await client.call_tool("power", {"a": 2, "b": 3})
assert power_result.structured_content is not None
assert power_result.structured_content["result"] == 8 # type: ignore[attr-defined]
assert isinstance(power_result.structured_content, dict)
assert power_result.structured_content["result"] == 8
modulo_result = await client.call_tool("modulo", {"a": 10, "b": 3})
assert modulo_result.structured_content is not None
assert modulo_result.structured_content["result"] == 1 # type: ignore[attr-defined]
assert isinstance(modulo_result.structured_content, dict)
assert modulo_result.structured_content["result"] == 1
async def test_injected_tool_with_complex_return_type(self, base_server: FastMCP):
"""Test injected tools with complex return types."""
@ -270,7 +274,8 @@ class TestToolInjectionMiddleware:
assert "add" in tool_names
assert "subtract" in tool_names
assert result.structured_content is not None
assert result.structured_content["result"] == 7 # type: ignore[attr-defined]
assert isinstance(result.structured_content, dict)
assert result.structured_content["result"] == 7
class TestPromptToolMiddleware:

View file

@ -179,7 +179,8 @@ class TestResourceOperations:
async with Client(main) as client:
result = await client.read_resource("resource://data")
assert result[0].text == "content" # type: ignore[attr-defined]
assert isinstance(result[0], mt.TextResourceContents)
assert result[0].text == "content"
class TestResourceTemplateOperations:
@ -225,7 +226,8 @@ class TestResourceTemplateOperations:
async with Client(main) as client:
result = await client.read_resource("resource://123/data")
assert result[0].text == "data for 123" # type: ignore[attr-defined]
assert isinstance(result[0], mt.TextResourceContents)
assert result[0].text == "data for 123"
class TestPromptOperations:
@ -277,7 +279,8 @@ class TestPromptOperations:
async with Client(main) as client:
result = await client.get_prompt("greet", {"name": "World"})
assert result.messages[0].content.text == "Hello, World!" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, mt.TextContent)
assert result.messages[0].content.text == "Hello, World!"
class TestServerReference:
@ -359,7 +362,8 @@ class TestMiddlewareChain:
async with Client(parent) as client:
result = await client.read_resource("data://c/gc/value")
assert result[0].text == "result" # type: ignore[attr-defined]
assert isinstance(result[0], mt.TextResourceContents)
assert result[0].text == "result"
assert calls == [
"parent:before",
@ -394,7 +398,8 @@ class TestMiddlewareChain:
async with Client(parent) as client:
result = await client.get_prompt("c_gc_greet", {"name": "World"})
assert result.messages[0].content.text == "Hello, World!" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, mt.TextContent)
assert result.messages[0].content.text == "Hello, World!"
assert calls == [
"parent:before",
@ -429,7 +434,8 @@ class TestMiddlewareChain:
async with Client(parent) as client:
result = await client.read_resource("item://c/gc/42")
assert result[0].text == "item-42" # type: ignore[attr-defined]
assert isinstance(result[0], mt.TextResourceContents)
assert result[0].text == "item-42"
assert calls == [
"parent:before",

View file

@ -16,6 +16,7 @@ from fastmcp.client.elicitation import ElicitRequestParams, ElicitResult
from fastmcp.client.logging import LogMessage
from fastmcp.client.sampling import RequestContext, SamplingMessage, SamplingParams
from fastmcp.exceptions import ToolError
from fastmcp.server.elicitation import AcceptedElicitation
from fastmcp.server.proxy import ProxyClient
@ -57,7 +58,9 @@ def fastmcp_server():
)
if result.action == "accept":
return f"Hello, {result.data.name}!" # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, Person)
return f"Hello, {result.data.name}!"
else:
return "No name provided."
@ -368,7 +371,9 @@ class TestProxyClient:
)
if result.action == "accept":
return f"Content: {result.data.content}, Acknowledge: {result.data.acknowledge}" # type: ignore[attr-defined]
assert isinstance(result, AcceptedElicitation)
assert isinstance(result.data, TestModel)
return f"Content: {result.data.content}, Acknowledge: {result.data.acknowledge}"
else:
return f"Elicitation {result.action}"

View file

@ -6,7 +6,7 @@ import pytest
from anyio import create_task_group
from dirty_equals import Contains
from mcp import McpError
from mcp.types import Icon
from mcp.types import Icon, TextContent, TextResourceContents
from pydantic import AnyUrl
from fastmcp import FastMCP
@ -132,7 +132,7 @@ def test_as_proxy_with_url():
assert isinstance(proxy, FastMCPProxy)
client = cast(Client, proxy.client_factory())
assert isinstance(client.transport, StreamableHttpTransport)
assert client.transport.url == "http://example.com/mcp/" # type: ignore[attr-defined]
assert client.transport.url == "http://example.com/mcp/"
async def test_proxy_with_async_client_factory():
@ -147,7 +147,7 @@ async def test_proxy_with_async_client_factory():
client = await proxy.client_factory() # type: ignore[misc]
assert isinstance(client, Client)
assert isinstance(client.transport, StreamableHttpTransport)
assert client.transport.url == "http://example.com/mcp/" # type: ignore[attr-defined]
assert client.transport.url == "http://example.com/mcp/"
class TestTools:
@ -231,7 +231,8 @@ class TestTools:
async with Client(proxy_server) as client:
result = await client.call_tool("tool_with_meta", {"value": "test"})
assert result.content[0].text == "Result: test" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Result: test"
assert result.meta == {"custom_key": "custom_value", "processed": True}
async def test_proxy_can_overwrite_proxied_tool(self, proxy_server):
@ -315,7 +316,8 @@ class TestResources:
async def test_read_resource(self, proxy_server: FastMCPProxy):
async with Client(proxy_server) as client:
result = await client.read_resource("resource://wave")
assert result[0].text == "👋" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "👋"
async def test_read_resource_same_as_original(self, fastmcp_server, proxy_server):
async with Client(fastmcp_server) as client:
@ -327,7 +329,8 @@ class TestResources:
async def test_read_json_resource(self, proxy_server: FastMCPProxy):
async with Client(proxy_server) as client:
result = await client.read_resource("data://users")
assert json.loads(result[0].text) == USERS # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert json.loads(result[0].text) == USERS
async def test_read_resource_returns_none_if_not_found(self, proxy_server):
with pytest.raises(
@ -347,7 +350,8 @@ class TestResources:
async with Client(proxy_server) as client:
result = await client.read_resource("resource://wave")
assert result[0].text == "Overwritten wave! 🌊" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Overwritten wave! 🌊"
async def test_proxy_errors_if_overwritten_resource_is_disabled(self, proxy_server):
"""
@ -420,7 +424,8 @@ class TestResourceTemplates:
async def test_read_resource_template(self, proxy_server: FastMCPProxy, id: int):
async with Client(proxy_server) as client:
result = await client.read_resource(f"data://user/{id}")
assert json.loads(result[0].text) == USERS[id - 1] # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert json.loads(result[0].text) == USERS[id - 1]
async def test_read_resource_template_same_as_original(
self, fastmcp_server, proxy_server
@ -447,7 +452,8 @@ class TestResourceTemplates:
async with Client(proxy_server) as client:
result = await client.read_resource("data://user/1")
user_data = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
user_data = json.loads(result[0].text)
assert user_data["name"] == "Overwritten User"
assert user_data["extra"] == "data"
@ -537,7 +543,8 @@ class TestPrompts:
async with Client(proxy_server) as client:
result = await client.get_prompt("welcome", {"name": "Alice"})
assert result.messages[0].role == "user"
assert result.messages[0].content.text == "Welcome to FastMCP, Alice!" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Welcome to FastMCP, Alice!"
async def test_proxy_can_overwrite_proxied_prompt(self, proxy_server):
"""
@ -553,8 +560,9 @@ class TestPrompts:
"welcome", {"name": "Alice", "extra": "colleague"}
)
assert result.messages[0].role == "user"
assert isinstance(result.messages[0].content, TextContent)
assert (
result.messages[0].content.text # type: ignore[attr-defined]
result.messages[0].content.text
== "Overwritten welcome, Alice! You are my colleague."
)

View file

@ -8,11 +8,14 @@ Tests that the server correctly enforces task execution modes:
import pytest
from mcp.shared.exceptions import McpError
from mcp.types import TextContent, ToolExecution
from mcp.types import Tool as MCPTool
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.exceptions import ToolError
from fastmcp.server.tasks import TaskConfig
from fastmcp.tools.tool import Tool
class TestTaskConfigNormalization:
@ -27,8 +30,8 @@ class TestTaskConfigNormalization:
return "ok"
tool = await mcp._tool_manager.get_tool("my_tool")
assert tool is not None
assert tool.task_config.mode == "optional" # type: ignore[attr-defined]
assert isinstance(tool, Tool)
assert tool.task_config.mode == "optional"
async def test_task_false_normalizes_to_forbidden(self):
"""task=False should normalize to TaskConfig(mode='forbidden')."""
@ -39,8 +42,8 @@ class TestTaskConfigNormalization:
return "ok"
tool = await mcp._tool_manager.get_tool("my_tool")
assert tool is not None
assert tool.task_config.mode == "forbidden" # type: ignore[attr-defined]
assert isinstance(tool, Tool)
assert tool.task_config.mode == "forbidden"
async def test_task_config_passed_directly(self):
"""TaskConfig should be preserved when passed directly."""
@ -51,8 +54,8 @@ class TestTaskConfigNormalization:
return "ok"
tool = await mcp._tool_manager.get_tool("my_tool")
assert tool is not None
assert tool.task_config.mode == "required" # type: ignore[attr-defined]
assert isinstance(tool, Tool)
assert tool.task_config.mode == "required"
async def test_default_task_inherits_server_default(self):
"""Default task value should inherit from server default."""
@ -64,8 +67,8 @@ class TestTaskConfigNormalization:
return "ok"
tool = await mcp_no_tasks._tool_manager.get_tool("my_tool_sync")
assert tool is not None
assert tool.task_config.mode == "forbidden" # type: ignore[attr-defined]
assert isinstance(tool, Tool)
assert tool.task_config.mode == "forbidden"
# Server with tasks enabled
mcp_tasks = FastMCP("test", tasks=True)
@ -75,8 +78,8 @@ class TestTaskConfigNormalization:
return "ok"
tool2 = await mcp_tasks._tool_manager.get_tool("my_tool_async")
assert tool2 is not None
assert tool2.task_config.mode == "optional" # type: ignore[attr-defined]
assert isinstance(tool2, Tool)
assert tool2.task_config.mode == "optional"
class TestToolModeEnforcement:
@ -254,7 +257,8 @@ class TestPromptModeEnforcement:
"""Forbidden mode succeeds when called without task metadata."""
async with Client(server) as client:
result = await client.get_prompt("forbidden_prompt")
assert "forbidden message" in str(result.messages[0].content) # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert "forbidden message" in str(result.messages[0].content)
class TestToolExecutionMetadata:
@ -271,8 +275,9 @@ class TestToolExecutionMetadata:
async with Client(mcp) as client:
tools = await client.list_tools()
tool = next(t for t in tools if t.name == "my_tool")
assert tool.execution is not None
assert tool.execution.taskSupport == "optional" # type: ignore[attr-defined]
assert isinstance(tool, MCPTool)
assert isinstance(tool.execution, ToolExecution)
assert tool.execution.taskSupport == "optional"
async def test_required_tool_exposes_task_support(self):
"""Tools with mode=required should expose taskSupport='required'."""
@ -285,8 +290,9 @@ class TestToolExecutionMetadata:
async with Client(mcp) as client:
tools = await client.list_tools()
tool = next(t for t in tools if t.name == "my_tool")
assert tool.execution is not None
assert tool.execution.taskSupport == "required" # type: ignore[attr-defined]
assert isinstance(tool, MCPTool)
assert isinstance(tool.execution, ToolExecution)
assert tool.execution.taskSupport == "required"
async def test_forbidden_tool_has_no_execution(self):
"""Tools with mode=forbidden should not expose execution metadata."""
@ -344,5 +350,5 @@ class TestSyncFunctionValidation:
return "ok"
tool = await mcp._tool_manager.get_tool("sync_tool")
assert tool is not None
assert tool.task_config.mode == "forbidden" # type: ignore[attr-defined]
assert isinstance(tool, Tool)
assert tool.task_config.mode == "forbidden"

View file

@ -12,6 +12,7 @@ Clients connecting through proxies can:
import pytest
from mcp.shared.exceptions import McpError
from mcp.types import TextContent, TextResourceContents
from fastmcp import FastMCP
from fastmcp.client import Client
@ -114,7 +115,8 @@ class TestProxyPromptsSyncExecution:
"""Prompt called without task=True works through proxy."""
async with Client(proxy_server) as client:
result = await client.get_prompt("greeting_prompt", {"name": "Alice"})
assert "Hello, Alice!" in result.messages[0].content.text # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert "Hello, Alice!" in result.messages[0].content.text
class TestProxyPromptsTaskForbidden:
@ -136,13 +138,15 @@ class TestProxyResourcesSyncExecution:
"""Resource read without task=True works through proxy."""
async with Client(proxy_server) as client:
result = await client.read_resource("data://info.txt")
assert "Important information from the backend" in result[0].text # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert "Important information from the backend" in result[0].text
async def test_resource_template_sync_execution_works(self, proxy_server: FastMCP):
"""Resource template without task=True works through proxy."""
async with Client(proxy_server) as client:
result = await client.read_resource("data://user/42.json")
assert '"id": "42"' in result[0].text # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert '"id": "42"' in result[0].text
class TestProxyResourcesTaskForbidden:

View file

@ -755,7 +755,8 @@ async def test_toolerror_propagates_from_dependency(mcp: FastMCP):
result = await client.call_tool("my_tool", {}, raise_on_error=False)
assert result.is_error
# The original error message should be preserved (not wrapped in RuntimeError)
assert result.content[0].text == "Client ID is required - select a client first" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Client ID is required - select a client first"
async def test_validation_error_propagates_from_dependency(mcp: FastMCP):
@ -776,4 +777,5 @@ async def test_validation_error_propagates_from_dependency(mcp: FastMCP):
"tool_with_validation", {}, raise_on_error=False
)
assert result.is_error
assert result.content[0].text == "Invalid input format" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Invalid input format"

View file

@ -233,5 +233,6 @@ class TestEventStoreIntegration:
await event_store.replay_events_after(event_id, callback)
assert len(replayed) == 1
assert replayed[0].message.root.method == "tools/call" # type: ignore[attr-defined]
assert replayed[0].message.root.id == "request-456" # type: ignore[attr-defined]
assert isinstance(replayed[0].message.root, JSONRPCRequest)
assert replayed[0].message.root.method == "tools/call"
assert replayed[0].message.root.id == "request-456"

View file

@ -9,6 +9,7 @@ strict_input_validation=False, the default).
import json
import pytest
from mcp.types import TextContent
from pydantic import BaseModel
from fastmcp import Client, FastMCP
@ -59,7 +60,8 @@ class TestStringToIntegerCoercion:
async with Client(mcp) as client:
# String integers should be coerced to integers
result = await client.call_tool("add_numbers", {"a": "10", "b": "20"})
assert result.content[0].text == "30" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "30"
async def test_default_is_not_strict(self):
"""By default, strict_input_validation should be False."""
@ -73,7 +75,8 @@ class TestStringToIntegerCoercion:
async with Client(mcp) as client:
# Should work with string integers by default
result = await client.call_tool("multiply", {"x": "5", "y": "3"})
assert result.content[0].text == "15" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "15"
async def test_string_float_coercion(self):
"""Test that string floats are also coerced."""
@ -88,7 +91,8 @@ class TestStringToIntegerCoercion:
result = await client.call_tool(
"calculate_area", {"length": "10.5", "width": "20.0"}
)
assert result.content[0].text == "210.0" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "210.0"
async def test_invalid_coercion_still_fails(self):
"""Even without strict validation, truly invalid inputs should fail."""
@ -122,8 +126,9 @@ class TestPydanticModelArguments:
"create_user",
{"profile": {"name": "Alice", "age": 30, "email": "alice@example.com"}},
)
assert "Alice" in result.content[0].text # type: ignore[attr-defined]
assert "30" in result.content[0].text # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "Alice" in result.content[0].text
assert "30" in result.content[0].text
async def test_pydantic_model_with_stringified_json_no_strict(self):
"""Test if stringified JSON is accepted for Pydantic models without strict validation."""
@ -144,7 +149,8 @@ class TestPydanticModelArguments:
try:
result = await client.call_tool("create_user", {"profile": stringified})
# If this succeeds, we're handling stringified JSON
assert "Bob" in result.content[0].text # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "Bob" in result.content[0].text
stringified_json_works = True
except Exception as e:
# If this fails, we're not handling stringified JSON
@ -182,8 +188,9 @@ class TestPydanticModelArguments:
}
},
)
assert "Charlie" in result.content[0].text # type: ignore[attr-defined]
assert "35" in result.content[0].text # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "Charlie" in result.content[0].text
assert "35" in result.content[0].text
async def test_pydantic_model_strict_validation(self):
"""With strict validation, Pydantic models should enforce exact types."""
@ -292,7 +299,8 @@ class TestEdgeCases:
result = await client.call_tool(
"format_message", {"text": "hi", "repeat": "3"}
)
assert result.content[0].text == "hihihi" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "hihihi"
async def test_none_values(self):
"""Test handling of None values."""
@ -305,7 +313,8 @@ class TestEdgeCases:
async with Client(mcp) as client:
result = await client.call_tool("process_optional", {"value": None})
assert "None" in result.content[0].text # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "None" in result.content[0].text
async def test_empty_string_to_int(self):
"""Empty strings should fail conversion to int."""
@ -332,11 +341,13 @@ class TestEdgeCases:
async with Client(mcp) as client:
# String "true" should be coerced to boolean
result = await client.call_tool("toggle", {"enabled": "true"})
assert "enabled" in result.content[0].text.lower() # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "enabled" in result.content[0].text.lower()
# String "false" should be coerced to boolean
result = await client.call_tool("toggle", {"enabled": "false"})
assert "disabled" in result.content[0].text.lower() # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "disabled" in result.content[0].text.lower()
async def test_list_of_integers_with_string_elements(self):
"""Test lists containing string representations of integers."""
@ -350,4 +361,5 @@ class TestEdgeCases:
async with Client(mcp) as client:
# List with string integers
result = await client.call_tool("sum_numbers", {"numbers": ["1", "2", "3"]})
assert result.content[0].text == "6" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "6"

View file

@ -3,6 +3,7 @@ import sys
from contextlib import asynccontextmanager
import pytest
from mcp.types import TextContent, TextResourceContents
from fastmcp import FastMCP
from fastmcp.client import Client
@ -144,7 +145,8 @@ class TestBasicMount:
# Test actual functionality
async with Client(main_app) as client:
resource_result = await client.read_resource("data://config")
assert resource_result[0].text == "Sub resource data" # type: ignore[attr-defined]
assert isinstance(resource_result[0], TextResourceContents)
assert resource_result[0].text == "Sub resource data"
async def test_mount_resource_templates_no_prefix(self):
"""Test mounting a server with resource templates without prefix."""
@ -165,7 +167,8 @@ class TestBasicMount:
# Test actual functionality
async with Client(main_app) as client:
template_result = await client.read_resource("users://123/info")
assert template_result[0].text == "Sub template for user 123" # type: ignore[attr-defined]
assert isinstance(template_result[0], TextResourceContents)
assert template_result[0].text == "Sub template for user 123"
async def test_mount_prompts_no_prefix(self):
"""Test mounting a server with prompts without prefix."""
@ -413,7 +416,8 @@ class TestPrefixConflictResolution:
# Test that reading the resource uses the first server's implementation
result = await client.read_resource("shared://data")
assert result[0].text == "First app data" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "First app data"
async def test_first_server_wins_resources_same_prefix(self):
"""Test that first mounted server wins for resources when same prefix is used."""
@ -444,7 +448,8 @@ class TestPrefixConflictResolution:
# Test that reading the resource uses the first server's implementation
result = await client.read_resource("shared://api/data")
assert result[0].text == "First app data" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "First app data"
async def test_first_server_wins_resource_templates_no_prefix(self):
"""Test that first mounted server wins for resource templates when no prefix is used."""
@ -475,7 +480,8 @@ class TestPrefixConflictResolution:
# Test that reading the resource uses the first server's implementation
result = await client.read_resource("users://123/profile")
assert result[0].text == "First app user 123" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "First app user 123"
async def test_first_server_wins_resource_templates_same_prefix(self):
"""Test that first mounted server wins for resource templates when same prefix is used."""
@ -506,7 +512,8 @@ class TestPrefixConflictResolution:
# Test that reading the resource uses the first server's implementation
result = await client.read_resource("users://api/123/profile")
assert result[0].text == "First app user 123" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "First app user 123"
async def test_first_server_wins_prompts_no_prefix(self):
"""Test that first mounted server wins for prompts when no prefix is used."""
@ -536,7 +543,8 @@ class TestPrefixConflictResolution:
# Test that getting the prompt uses the first server's implementation
result = await client.get_prompt("shared_prompt", {})
assert result.messages is not None
assert result.messages[0].content.text == "First app prompt" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "First app prompt"
async def test_first_server_wins_prompts_same_prefix(self):
"""Test that first mounted server wins for prompts when same prefix is used."""
@ -568,7 +576,8 @@ class TestPrefixConflictResolution:
# Test that getting the prompt uses the first server's implementation
result = await client.get_prompt("api_shared_prompt", {})
assert result.messages is not None
assert result.messages[0].content.text == "First app prompt" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "First app prompt"
class TestDynamicChanges:
@ -646,7 +655,8 @@ class TestResourcesAndTemplates:
# Check that resource can be accessed
async with Client(main_app) as client:
result = await client.read_resource("data://data/users")
assert json.loads(result[0].text) == ["user1", "user2"] # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert json.loads(result[0].text) == ["user1", "user2"]
async def test_mount_with_resource_templates(self):
"""Test mounting a server with resource templates."""
@ -667,7 +677,8 @@ class TestResourcesAndTemplates:
# Check template instantiation
async with Client(main_app) as client:
result = await client.read_resource("users://api/123/profile")
profile = json.loads(result[0].text) # type: ignore
assert isinstance(result[0], TextResourceContents)
profile = json.loads(result[0].text)
assert profile["id"] == "123"
assert profile["name"] == "User 123"
@ -691,7 +702,8 @@ class TestResourcesAndTemplates:
# Check access to the resource
async with Client(main_app) as client:
result = await client.read_resource("data://data/config")
config = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
config = json.loads(result[0].text)
assert config["version"] == "1.0"
@ -817,7 +829,8 @@ class TestProxyServer:
# Resource should be accessible through main app
async with Client(main_app) as client:
result = await client.read_resource("config://proxy/settings")
config = json.loads(result[0].text) # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
config = json.loads(result[0].text)
assert config["api_key"] == "12345"
async def test_proxy_server_with_prompts(self):
@ -1148,7 +1161,8 @@ class TestCustomRouteForwarding:
routes = server._get_additional_http_routes()
assert len(routes) == 1
assert routes[0].path == "/test" # type: ignore[attr-defined]
assert hasattr(routes[0], "path")
assert routes[0].path == "/test"
async def test_mounted_servers_tracking(self):
"""Test that _providers list tracks mounted servers correctly."""
@ -1195,7 +1209,7 @@ class TestCustomRouteForwarding:
routes = server._get_additional_http_routes()
assert len(routes) == 2
route_paths = [route.path for route in routes] # type: ignore[attr-defined]
route_paths = [route.path for route in routes if hasattr(route, "path")]
assert "/route1" in route_paths
assert "/route2" in route_paths
@ -1306,13 +1320,15 @@ class TestDeeplyNestedMount:
async with Client(root) as client:
# Prompt at level 2 should work
result = await client.get_prompt("middle_middle_prompt", {"name": "World"})
assert "Hello from middle: World" in result.messages[0].content.text # type: ignore[union-attr]
assert isinstance(result.messages[0].content, TextContent)
assert "Hello from middle: World" in result.messages[0].content.text
# Prompt at level 3 should also work
result = await client.get_prompt(
"middle_leaf_leaf_prompt", {"name": "Test"}
)
assert "Hello from leaf: Test" in result.messages[0].content.text # type: ignore[union-attr]
assert isinstance(result.messages[0].content, TextContent)
assert "Hello from leaf: Test" in result.messages[0].content.text
async def test_four_level_nested_tool_invocation(self):
"""Test invoking tools from servers mounted 4 levels deep."""

View file

@ -4,7 +4,7 @@ from collections.abc import Sequence
from typing import Any
import pytest
from mcp.types import AnyUrl
from mcp.types import AnyUrl, TextContent
from mcp.types import Tool as MCPTool
from fastmcp import FastMCP
@ -169,8 +169,9 @@ class TestProvider:
)
assert result.structured_content is not None
assert result.structured_content["result"] == 42 # type: ignore[attr-defined]
assert result.structured_content["operation"] == "multiply" # type: ignore[attr-defined]
assert isinstance(result.structured_content, dict)
assert result.structured_content["result"] == 42
assert result.structured_content["operation"] == "multiply"
async def test_call_dynamic_tool_with_config(
self, base_server: FastMCP, dynamic_tools: list[Tool]
@ -186,7 +187,8 @@ class TestProvider:
assert result.structured_content is not None
# 5 + 3 + 100 (value offset) = 108
assert result.structured_content["result"] == 108 # type: ignore[attr-defined]
assert isinstance(result.structured_content, dict)
assert result.structured_content["result"] == 108
async def test_call_static_tool_still_works(
self, base_server: FastMCP, dynamic_tools: list[Tool]
@ -201,7 +203,8 @@ class TestProvider:
)
assert result.structured_content is not None
assert result.structured_content["result"] == 15 # type: ignore[attr-defined]
assert isinstance(result.structured_content, dict)
assert result.structured_content["result"] == 15
async def test_call_tool_uses_get_tool_for_efficient_lookup(
self, base_server: FastMCP, dynamic_tools: list[Tool]
@ -280,7 +283,8 @@ class TestProvider:
)
assert result.structured_content is not None
assert result.structured_content["result"] == 7 # type: ignore[attr-defined]
assert isinstance(result.structured_content, dict)
assert result.structured_content["result"] == 7
class TestProviderClass:
@ -379,7 +383,8 @@ class TestProviderExecutionMethods:
result = await client.call_tool("test_tool", {"a": 1, "b": 2})
assert result.structured_content is not None
assert result.structured_content["result"] == 3 # type: ignore[attr-defined]
assert isinstance(result.structured_content, dict)
assert result.structured_content["result"] == 3
async def test_read_resource_default_implementation(self):
"""Test that default read_resource uses get_resource and reads it."""
@ -448,4 +453,5 @@ class TestProviderExecutionMethods:
result = await client.get_prompt("greeting", {"name": "World"})
assert len(result.messages) == 1
assert result.messages[0].content.text == "Hello, World!" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Hello, World!"

View file

@ -5,11 +5,12 @@ from typing import Annotated
import pytest
from mcp import McpError
from mcp.types import BlobResourceContents, TextContent, TextResourceContents
from pydantic import Field
from fastmcp import Client, FastMCP
from fastmcp.exceptions import NotFoundError
from fastmcp.prompts.prompt import FunctionPrompt, Prompt
from fastmcp.prompts.prompt import FunctionPrompt, Prompt, PromptResult
from fastmcp.resources import Resource, ResourceContent, ResourceTemplate
from fastmcp.tools import FunctionTool
from fastmcp.tools.tool import Tool
@ -451,7 +452,8 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert result[0].text == "Hello, world!" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Hello, world!"
async def test_resource_decorator_incorrect_usage(self):
mcp = FastMCP()
@ -478,7 +480,8 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert result[0].text == "Hello, world!" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Hello, world!"
async def test_resource_decorator_with_description(self):
mcp = FastMCP()
@ -525,7 +528,8 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert result[0].text == "My prefix: Hello, world!" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "My prefix: Hello, world!"
async def test_resource_decorator_classmethod(self):
mcp = FastMCP()
@ -545,7 +549,8 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert result[0].text == "Class prefix: Hello, world!" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Class prefix: Hello, world!"
async def test_resource_decorator_classmethod_error(self):
mcp = FastMCP()
@ -569,7 +574,8 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert result[0].text == "Static Hello, world!" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Static Hello, world!"
async def test_resource_decorator_async_function(self):
mcp = FastMCP()
@ -580,7 +586,8 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert result[0].text == "Async Hello, world!" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Async Hello, world!"
async def test_resource_decorator_staticmethod_order(self):
"""Test that both decorator orders work for static methods"""
@ -594,7 +601,8 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert result[0].text == "Static Hello, world!" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Static Hello, world!"
async def test_resource_decorator_with_meta(self):
"""Test that meta parameter is passed through the resource decorator."""
@ -626,10 +634,12 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://widget")
assert len(result) == 1
assert result[0].text == "<widget>content</widget>" # type: ignore[attr-defined]
assert result[0].mimeType == "text/html" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "<widget>content</widget>"
assert result[0].mimeType == "text/html"
# Meta should be in the response
assert result[0].meta == {"csp": "script-src 'self'", "version": "1.0"} # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].meta == {"csp": "script-src 'self'", "version": "1.0"}
async def test_resource_content_binary_with_meta(self):
"""Test that ResourceContent with binary content and meta works."""
@ -647,7 +657,8 @@ class TestResourceDecorator:
assert len(result) == 1
# Binary content comes back as blob
assert hasattr(result[0], "blob")
assert result[0].meta == {"encoding": "raw"} # type: ignore[attr-defined]
assert isinstance(result[0], BlobResourceContents)
assert result[0].meta == {"encoding": "raw"}
async def test_resource_content_without_meta(self):
"""Test that ResourceContent without meta works (meta is None)."""
@ -660,9 +671,11 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://plain")
assert len(result) == 1
assert result[0].text == "plain content" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "plain content"
# Meta should be None
assert result[0].meta is None # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].meta is None
class TestTemplateDecorator:
@ -681,7 +694,8 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert result[0].text == "Data for test" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Data for test"
async def test_template_decorator_incorrect_usage(self):
mcp = FastMCP()
@ -708,7 +722,8 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert result[0].text == "Data for test" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Data for test"
async def test_template_decorator_with_description(self):
mcp = FastMCP()
@ -742,7 +757,8 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert result[0].text == "My prefix: Data for test" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "My prefix: Data for test"
async def test_template_decorator_classmethod(self):
mcp = FastMCP()
@ -763,7 +779,8 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert result[0].text == "Class prefix: Data for test" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Class prefix: Data for test"
async def test_template_decorator_staticmethod(self):
mcp = FastMCP()
@ -776,7 +793,8 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert result[0].text == "Static Data for test" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Static Data for test"
async def test_template_decorator_async_function(self):
mcp = FastMCP()
@ -787,7 +805,8 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert result[0].text == "Async Data for test" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Async Data for test"
async def test_template_decorator_with_tags(self):
"""Test that the template decorator properly sets tags."""
@ -843,7 +862,10 @@ class TestPromptDecorator:
assert prompt.name == "fn"
# Don't compare functions directly since validate_call wraps them
content = await prompt.render()
assert content.messages[0].content.text == "Hello, world!" # type: ignore[attr-defined]
if not isinstance(content, PromptResult):
content = PromptResult.from_value(content)
assert isinstance(content.messages[0].content, TextContent)
assert content.messages[0].content.text == "Hello, world!"
async def test_prompt_decorator_without_parentheses(self):
mcp = FastMCP()
@ -861,7 +883,8 @@ class TestPromptDecorator:
async with Client(mcp) as client:
result = await client.get_prompt("fn")
assert len(result.messages) == 1
assert result.messages[0].content.text == "Hello, world!" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Hello, world!"
async def test_prompt_decorator_with_name(self):
mcp = FastMCP()
@ -875,7 +898,10 @@ class TestPromptDecorator:
prompt = prompts_dict["custom_name"]
assert prompt.name == "custom_name"
content = await prompt.render()
assert content.messages[0].content.text == "Hello, world!" # type: ignore[attr-defined]
if not isinstance(content, PromptResult):
content = PromptResult.from_value(content)
assert isinstance(content.messages[0].content, TextContent)
assert content.messages[0].content.text == "Hello, world!"
async def test_prompt_decorator_with_description(self):
mcp = FastMCP()
@ -889,7 +915,10 @@ class TestPromptDecorator:
prompt = prompts_dict["fn"]
assert prompt.description == "A custom description"
content = await prompt.render()
assert content.messages[0].content.text == "Hello, world!" # type: ignore[attr-defined]
if not isinstance(content, PromptResult):
content = PromptResult.from_value(content)
assert isinstance(content.messages[0].content, TextContent)
assert content.messages[0].content.text == "Hello, world!"
async def test_prompt_decorator_with_parameters(self):
mcp = FastMCP()
@ -912,14 +941,16 @@ class TestPromptDecorator:
result = await client.get_prompt("test_prompt", {"name": "World"})
assert len(result.messages) == 1
message = result.messages[0]
assert message.content.text == "Hello, World!" # type: ignore[attr-defined]
assert isinstance(message.content, TextContent)
assert message.content.text == "Hello, World!"
result = await client.get_prompt(
"test_prompt", {"name": "World", "greeting": "Hi"}
)
assert len(result.messages) == 1
message = result.messages[0]
assert message.content.text == "Hi, World!" # type: ignore[attr-defined]
assert isinstance(message.content, TextContent)
assert message.content.text == "Hi, World!"
async def test_prompt_decorator_instance_method(self):
mcp = FastMCP()
@ -938,7 +969,8 @@ class TestPromptDecorator:
result = await client.get_prompt("test_prompt")
assert len(result.messages) == 1
message = result.messages[0]
assert message.content.text == "My prefix: Hello, world!" # type: ignore[attr-defined]
assert isinstance(message.content, TextContent)
assert message.content.text == "My prefix: Hello, world!"
async def test_prompt_decorator_classmethod(self):
mcp = FastMCP()
@ -956,7 +988,8 @@ class TestPromptDecorator:
result = await client.get_prompt("test_prompt")
assert len(result.messages) == 1
message = result.messages[0]
assert message.content.text == "Class prefix: Hello, world!" # type: ignore[attr-defined]
assert isinstance(message.content, TextContent)
assert message.content.text == "Class prefix: Hello, world!"
async def test_prompt_decorator_classmethod_error(self):
mcp = FastMCP()
@ -982,7 +1015,8 @@ class TestPromptDecorator:
result = await client.get_prompt("test_prompt")
assert len(result.messages) == 1
message = result.messages[0]
assert message.content.text == "Static Hello, world!" # type: ignore[attr-defined]
assert isinstance(message.content, TextContent)
assert message.content.text == "Static Hello, world!"
async def test_prompt_decorator_async_function(self):
mcp = FastMCP()
@ -995,7 +1029,8 @@ class TestPromptDecorator:
result = await client.get_prompt("test_prompt")
assert len(result.messages) == 1
message = result.messages[0]
assert message.content.text == "Async Hello, world!" # type: ignore[attr-defined]
assert isinstance(message.content, TextContent)
assert message.content.text == "Async Hello, world!"
async def test_prompt_decorator_with_tags(self):
"""Test that the prompt decorator properly sets tags."""
@ -1028,7 +1063,8 @@ class TestPromptDecorator:
async with Client(mcp) as client:
result = await client.get_prompt("string_named_prompt")
assert len(result.messages) == 1
assert result.messages[0].content.text == "Hello from string named prompt!" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Hello from string named prompt!"
async def test_prompt_direct_function_call(self):
"""Test that prompts can be registered via direct function call."""
@ -1052,7 +1088,8 @@ class TestPromptDecorator:
async with Client(mcp) as client:
result = await client.get_prompt("direct_call_prompt")
assert len(result.messages) == 1
assert result.messages[0].content.text == "Hello from direct call!" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Hello from direct call!"
async def test_prompt_decorator_conflicting_names_error(self):
"""Test that providing both positional and keyword names raises an error."""
@ -1081,7 +1118,8 @@ class TestPromptDecorator:
result = await client.get_prompt("test_prompt")
assert len(result.messages) == 1
message = result.messages[0]
assert message.content.text == "Static Hello, world!" # type: ignore[attr-defined]
assert isinstance(message.content, TextContent)
assert message.content.text == "Static Hello, world!"
async def test_prompt_decorator_with_meta(self):
"""Test that meta parameter is passed through the prompt decorator."""
@ -1135,17 +1173,20 @@ class TestResourcePrefixMounting:
async with Client(main_server) as client:
# Regular resource
result = await client.read_resource("resource://prefix/test-resource")
assert result[0].text == "Resource content" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Resource content"
# Absolute path resource
result = await client.read_resource("resource://prefix//absolute/path")
assert result[0].text == "Absolute resource content" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Absolute resource content"
# Template resource
result = await client.read_resource(
"resource://prefix/param-value/template"
)
assert result[0].text == "Template resource with param-value" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource with param-value"
class TestShouldIncludeComponent:
@ -1344,4 +1385,5 @@ class TestAbstractCollectionTypes:
# Verify it works with a client
async with Client(mcp) as client:
result = await client.call_tool("greet", {"name": "World"})
assert result.content[0].text == "Hello, World!" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Hello, World!"

View file

@ -26,7 +26,7 @@ from fastmcp import Client, Context, FastMCP
from fastmcp.client.client import CallToolResult
from fastmcp.client.transports import FastMCPTransport
from fastmcp.exceptions import ToolError
from fastmcp.prompts.prompt import Prompt, PromptMessage
from fastmcp.prompts.prompt import Prompt, PromptMessage, PromptResult
from fastmcp.resources import FileResource, ResourceTemplate
from fastmcp.resources.resource import FunctionResource
from fastmcp.tools.tool import Tool, ToolResult
@ -156,13 +156,15 @@ class TestTools:
async def test_call_tool_mcp(self, tool_server: FastMCP):
async with Client(tool_server) as client:
result = await client.call_tool_mcp("add", {"x": 1, "y": 2})
assert result.content[0].text == "3" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "3"
assert result.structuredContent == {"result": 3}
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 result.content[0].text == "3" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "3"
assert result.structured_content == {"result": 3}
assert result.data == 3
@ -190,7 +192,8 @@ class TestTools:
result = await client.call_tool("list_tool", {})
# Adjacent non-MCP list items are combined into single content block
assert len(result.content) == 1
assert result.content[0].text == '["x",2]' # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == '["x",2]'
assert result.data == ["x", 2]
async def test_file_text_tool(self, tool_server: FastMCP):
@ -914,7 +917,7 @@ class TestToolOutputSchema:
mcp = FastMCP()
@mcp.tool
def f() -> annotation: # type: ignore
def f() -> annotation:
return "hello"
async with Client(mcp) as client:
@ -940,7 +943,7 @@ class TestToolOutputSchema:
mcp = FastMCP()
@mcp.tool
def f() -> annotation: # type: ignore[valid-type]
def f() -> annotation:
return {"name": "John", "age": 30}
async with Client(mcp) as client:
@ -966,7 +969,8 @@ class TestToolOutputSchema:
async with Client(mcp) as client:
result = await client.call_tool("f", {})
assert result.content[0].text == "42" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "42"
assert result.structured_content is None
assert result.data is None
@ -983,7 +987,8 @@ class TestToolOutputSchema:
async with Client(mcp) as client:
result = await client.call_tool("f", {})
assert result.content[0].text == "Hello, world!" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Hello, world!"
assert result.structured_content == {"message": "Hello, world!"}
assert result.data == {"message": "Hello, world!"}
@ -1007,7 +1012,8 @@ class TestToolOutputSchema:
result = await client.call_tool("simple_tool", {})
assert result.structured_content is None
assert result.data is None
assert result.content[0].text == "42" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "42"
async def test_output_schema_explicit_object_full_handshake(self):
"""Test explicit object output schema through full client/server handshake."""
@ -1044,6 +1050,8 @@ class TestToolOutputSchema:
result = await client.call_tool("explicit_tool", {})
assert result.structured_content == {"greeting": "Hello", "count": 42}
# Client deserializes according to schema, so check fields
# result.data is a dynamically generated Root type, so check attributes directly
assert result.data is not None
assert result.data.greeting == "Hello" # type: ignore[attr-defined]
assert result.data.count == 42 # type: ignore[attr-defined]
@ -1131,6 +1139,8 @@ class TestToolOutputSchema:
result = await client.call_tool("dataclass_tool", {})
assert result.structured_content == {"name": "Alice", "age": 30}
# Client deserializes according to schema
# result.data is a dynamically generated Root type, so check attributes directly
assert result.data is not None
assert result.data.name == "Alice" # type: ignore[attr-defined]
assert result.data.age == 30 # type: ignore[attr-defined]
@ -1450,7 +1460,8 @@ class TestResource:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test"))
assert result[0].text == "Hello, world!" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Hello, world!"
async def test_binary_resource(self):
mcp = FastMCP()
@ -1468,7 +1479,8 @@ class TestResource:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://binary"))
assert result[0].blob == base64.b64encode(b"Binary data").decode() # type: ignore[attr-defined]
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()
@ -1484,7 +1496,8 @@ class TestResource:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("file://test.txt"))
assert result[0].text == "Hello from file!" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Hello from file!"
async def test_file_resource_binary(self, tmp_path: Path):
mcp = FastMCP()
@ -1503,7 +1516,8 @@ class TestResource:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("file://test.bin"))
assert result[0].blob == base64.b64encode(b"Binary file data").decode() # type: ignore[attr-defined]
assert isinstance(result[0], BlobResourceContents)
assert result[0].blob == base64.b64encode(b"Binary file data").decode()
async def test_resource_with_annotations(self):
mcp = FastMCP()
@ -1587,7 +1601,8 @@ class TestResourceTags:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://1"))
assert result[0].text == "1" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "1"
with pytest.raises(McpError, match="Unknown resource"):
await client.read_resource(AnyUrl("resource://2"))
@ -1611,7 +1626,8 @@ class TestResourceContext:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test"))
assert result[0].text == "1" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "1"
class TestResourceEnabled:
@ -1756,7 +1772,8 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test/data"))
assert result[0].text == "Data for test" # type: ignore[attr-defined]
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"""
@ -1783,7 +1800,8 @@ class TestResourceTemplates:
result = await client.read_resource(
AnyUrl("resource://cursor/fastmcp/data")
)
assert result[0].text == "Data for cursor/fastmcp" # type: ignore[attr-defined]
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"""
@ -1807,7 +1825,8 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://static"))
assert result[0].text == "Static data" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Static data"
async def test_template_with_varkwargs(self):
"""Test that a template can have **kwargs."""
@ -1819,7 +1838,8 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("test://1/2/3"))
assert result[0].text == "6" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "6"
async def test_template_with_default_params(self):
"""Test that a template can have default parameters."""
@ -1838,11 +1858,13 @@ class TestResourceTemplates:
# 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 result[0].text == "15" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "15"
# Can also call with explicit params
result2 = await client.read_resource(AnyUrl("math://add/7"))
assert result2[0].text == "17" # type: ignore[attr-defined]
assert isinstance(result2[0], TextResourceContents)
assert result2[0].text == "17"
async def test_template_to_resource_conversion(self):
"""Test that a template can be converted to a resource."""
@ -1861,7 +1883,8 @@ class TestResourceTemplates:
# When accessed, should create a concrete resource
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test/data"))
assert result[0].text == "Data for test" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Data for test"
async def test_template_decorator_with_tags(self):
mcp = FastMCP()
@ -1883,7 +1906,8 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test/data"))
assert result[0].text == "Template resource: test/data" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource: test/data"
async def test_template_with_query_params(self):
"""Test RFC 6570 query parameters in resource templates."""
@ -1896,17 +1920,20 @@ class TestResourceTemplates:
async with Client(mcp) as client:
# No query params - uses defaults
result = await client.read_resource(AnyUrl("data://123"))
assert result[0].text == "id=123, format=json, limit=10" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "id=123, format=json, limit=10"
# One query param
result = await client.read_resource(AnyUrl("data://123?format=xml"))
assert result[0].text == "id=123, format=xml, limit=10" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "id=123, format=xml, limit=10"
# Multiple query params
result = await client.read_resource(
AnyUrl("data://123?format=csv&limit=50")
)
assert result[0].text == "id=123, format=csv, limit=50" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "id=123, format=csv, limit=50"
async def test_templates_match_in_order_of_definition(self):
"""
@ -1926,10 +1953,12 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://a/b/c"))
assert result[0].text == "Template resource 1: a/b/c" # type: ignore[attr-defined]
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 result[0].text == "Template resource 1: a/b" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource 1: a/b"
async def test_templates_shadow_each_other_reorder(self):
"""
@ -1948,10 +1977,12 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://a/b/c"))
assert result[0].text == "Template resource 2: a/b/c" # type: ignore[attr-defined]
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 result[0].text == "Template resource 1: a/b" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource 1: a/b"
async def test_resource_template_with_annotations(self):
"""Test that resource template annotations are visible to clients."""
@ -2035,7 +2066,8 @@ class TestResourceTemplatesTags:
async with Client(mcp) as client:
result = await client.read_resource("resource://1/x")
assert result[0].text == "Template resource 1: x" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource 1: x"
with pytest.raises(McpError, match="Unknown resource"):
await client.read_resource("resource://2/x")
@ -2048,7 +2080,8 @@ class TestResourceTemplatesTags:
await client.read_resource("resource://1/x")
result = await client.read_resource("resource://2/x")
assert result[0].text == "Template resource 2: x" # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource 2: x"
class TestResourceTemplateContext:
@ -2062,7 +2095,8 @@ class TestResourceTemplateContext:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test"))
assert result[0].text.startswith("Resource template: test 1") # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text.startswith("Resource template: test 1")
async def test_resource_template_context_with_callable_object(self):
mcp = FastMCP()
@ -2078,7 +2112,8 @@ class TestResourceTemplateContext:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test"))
assert result[0].text.startswith("Resource template: test 1") # type: ignore[attr-defined]
assert isinstance(result[0], TextResourceContents)
assert result[0].text.startswith("Resource template: test 1")
class TestResourceTemplateEnabled:
@ -2194,7 +2229,10 @@ class TestPrompts:
assert prompt.name == "fn"
# Don't compare functions directly since validate_call wraps them
content = await prompt.render()
assert content.messages[0].content.text == "Hello, world!" # type: ignore[attr-defined]
if not isinstance(content, PromptResult):
content = PromptResult.from_value(content)
assert isinstance(content.messages[0].content, TextContent)
assert content.messages[0].content.text == "Hello, world!"
async def test_prompt_decorator_with_name(self):
"""Test prompt decorator with custom name."""
@ -2209,7 +2247,10 @@ class TestPrompts:
prompt = prompts_dict["custom_name"]
assert prompt.name == "custom_name"
content = await prompt.render()
assert content.messages[0].content.text == "Hello, world!" # type: ignore[attr-defined]
if not isinstance(content, PromptResult):
content = PromptResult.from_value(content)
assert isinstance(content.messages[0].content, TextContent)
assert content.messages[0].content.text == "Hello, world!"
async def test_prompt_decorator_with_description(self):
"""Test prompt decorator with custom description."""
@ -2224,7 +2265,10 @@ class TestPrompts:
prompt = prompts_dict["fn"]
assert prompt.description == "A custom description"
content = await prompt.render()
assert content.messages[0].content.text == "Hello, world!" # type: ignore[attr-defined]
if not isinstance(content, PromptResult):
content = PromptResult.from_value(content)
assert isinstance(content.messages[0].content, TextContent)
assert content.messages[0].content.text == "Hello, world!"
async def test_prompt_decorator_with_parens(self):
mcp = FastMCP()
@ -2331,7 +2375,8 @@ class TestPrompts:
message = result.messages[0]
assert message.role == "user"
content = message.content
assert content.text == "Hello, World!" # type: ignore[attr-defined]
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."""
@ -2543,7 +2588,8 @@ class TestPromptContext:
assert len(result.messages) == 1
message = result.messages[0]
assert message.role == "user"
assert message.content.text == "Hello, World! 1" # type: ignore[attr-defined]
assert isinstance(message.content, TextContent)
assert message.content.text == "Hello, World! 1"
class TestPromptTags:
@ -2600,7 +2646,8 @@ class TestPromptTags:
async with Client(mcp) as client:
result = await client.get_prompt("prompt_1")
assert result.messages[0].content.text == "1" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "1"
with pytest.raises(McpError, match="Unknown prompt"):
await client.get_prompt("prompt_2")
@ -2613,7 +2660,8 @@ class TestPromptTags:
await client.get_prompt("prompt_1")
result = await client.get_prompt("prompt_2")
assert result.messages[0].content.text == "2" # type: ignore[attr-defined]
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "2"
class TestMeta:

View file

@ -1,6 +1,7 @@
from typing import Any
from mcp.types import ToolAnnotations
from mcp.types import Tool as MCPTool
from mcp.types import ToolAnnotations, ToolExecution
from fastmcp import Client, FastMCP
from fastmcp.tools.tool import Tool
@ -234,8 +235,9 @@ async def test_task_execution_auto_populated_for_task_enabled_tool():
tools_result = await client.list_tools()
assert len(tools_result) == 1
assert tools_result[0].name == "background_tool"
assert tools_result[0].execution is not None
assert tools_result[0].execution.taskSupport == "optional" # type: ignore[attr-defined]
assert isinstance(tools_result[0], MCPTool)
assert isinstance(tools_result[0].execution, ToolExecution)
assert tools_result[0].execution.taskSupport == "optional"
async def test_task_execution_omitted_for_task_disabled_tool():

View file

@ -741,7 +741,8 @@ class TestToolFromFunctionOutputSchema:
# Dict objects automatically become structured content even without schema
assert result.structured_content == {"message": "Hello, world!"}
assert len(result.content) == 1
assert result.content[0].text == '{"message":"Hello, world!"}' # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == '{"message":"Hello, world!"}'
async def test_output_schema_none_disables_structured_content(self):
"""Test that output_schema=None explicitly disables structured content."""
@ -755,7 +756,8 @@ class TestToolFromFunctionOutputSchema:
result = await tool.run({})
assert result.structured_content is None
assert len(result.content) == 1
assert result.content[0].text == "42" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "42"
async def test_output_schema_inferred_when_not_specified(self):
"""Test that output schema is inferred when not explicitly specified."""
@ -795,7 +797,8 @@ class TestToolFromFunctionOutputSchema:
result = await tool.run({})
# Dict result with object schema is used directly
assert result.structured_content == {"value": 42}
assert result.content[0].text == '{"value":42}' # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == '{"value":42}'
async def test_explicit_object_schema_with_non_dict_return_fails(self):
"""Test that explicit object schemas fail when function returns non-dict."""
@ -849,7 +852,8 @@ class TestToolFromFunctionOutputSchema:
result = await tool.run({})
# Unstructured content
assert len(result.content) == 1
assert result.content[0].text == "hello" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "hello"
# Structured content should be wrapped
assert result.structured_content == {"result": "hello"}
@ -917,7 +921,9 @@ class TestToolFromFunctionOutputSchema:
assert result_default.structured_content == {
"result": 123
} # Schema-based generation with wrapping
assert result_none.content[0].text == result_default.content[0].text == "123" # type: ignore[attr-defined]
assert isinstance(result_none.content[0], TextContent)
assert isinstance(result_default.content[0], TextContent)
assert result_none.content[0].text == result_default.content[0].text == "123"
async def test_non_object_output_schema_raises_error(self):
"""Test that providing a non-object output schema raises a ValueError."""

View file

@ -136,7 +136,9 @@ class TestAddTools:
def test_add_noncallable_tool(self):
manager = ToolManager()
with pytest.raises(TypeError, match="not a callable object"):
tool = Tool.from_function(1) # type: ignore
assert isinstance(1, int) # Intentionally passing invalid type
# Intentionally passing invalid type to test error handling
tool = Tool.from_function(1) # type: ignore[arg-type]
manager.add_tool(tool)
def test_add_lambda(self):
@ -407,7 +409,8 @@ class TestCallTools:
manager.add_tool(tool)
result = await manager.call_tool("add", {"a": 1, "b": 2})
assert result.content[0].text == "3" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "3"
assert result.structured_content == {"result": 3}
async def test_call_async_tool(self):
@ -419,7 +422,8 @@ class TestCallTools:
tool = Tool.from_function(double)
manager.add_tool(tool)
result = await manager.call_tool("double", {"n": 5})
assert result.content[0].text == "10" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "10"
assert result.structured_content == {"result": 10}
async def test_call_tool_callable_object(self):
@ -434,7 +438,8 @@ class TestCallTools:
tool = Tool.from_function(Adder())
manager.add_tool(tool)
result = await manager.call_tool("Adder", {"x": 1, "y": 2})
assert result.content[0].text == "3" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "3"
assert result.structured_content == {"result": 3}
async def test_call_tool_callable_object_async(self):
@ -449,7 +454,8 @@ class TestCallTools:
tool = Tool.from_function(Adder())
manager.add_tool(tool)
result = await manager.call_tool("Adder", {"x": 1, "y": 2})
assert result.content[0].text == "3" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "3"
assert result.structured_content == {"result": 3}
async def test_call_tool_with_default_args(self):
@ -462,7 +468,8 @@ class TestCallTools:
manager.add_tool(tool)
result = await manager.call_tool("add", {"a": 1})
assert result.content[0].text == "2" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "2"
assert result.structured_content == {"result": 2}
async def test_call_tool_with_missing_args(self):
@ -511,7 +518,8 @@ class TestCallTools:
result = await manager.call_tool(
"add_transformed", {"a_transformed": 1, "b_transformed": 2}
)
assert result.content[0].text == "3" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "3"
assert result.structured_content == {"result": 3}
async def test_call_tool_with_list_int_input(self):
@ -523,7 +531,8 @@ class TestCallTools:
manager.add_tool(tool)
result = await manager.call_tool("sum_vals", {"vals": [1, 2, 3]})
assert result.content[0].text == "6" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "6"
assert result.structured_content == {"result": 6}
async def test_call_tool_with_list_str_or_str_input(self):
@ -536,11 +545,13 @@ class TestCallTools:
# Try both with plain python object and with JSON list
result = await manager.call_tool("concat_strs", {"vals": ["a", "b", "c"]})
assert result.content[0].text == "abc" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "abc"
assert result.structured_content == {"result": "abc"}
result = await manager.call_tool("concat_strs", {"vals": "a"})
assert result.content[0].text == "a" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "a"
assert result.structured_content == {"result": "a"}
async def test_call_tool_with_complex_model(self):
@ -596,7 +607,8 @@ class TestCallTools:
return {"key": "value", "number": 123}
result = await manager.call_tool("get_data", {})
assert result.content[0].text == 'CUSTOM:{"key": "value", "number": 123}' # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == 'CUSTOM:{"key": "value", "number": 123}'
assert result.structured_content == {"key": "value", "number": 123}
async def test_call_tool_with_list_result_custom_serializer(self):
@ -653,10 +665,8 @@ class TestCallTools:
return uuid_result
result = await manager.call_tool("get_data", {})
assert (
result.content[0].text # type: ignore[attr-defined]
== pydantic_core.to_json(uuid_result).decode()
)
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == pydantic_core.to_json(uuid_result).decode()
assert result.structured_content == {"result": str(uuid_result)}
@ -727,7 +737,8 @@ class TestContextHandling:
async with context:
result = await manager.call_tool("tool_with_context", {"x": 42})
assert result.content[0].text == "42" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "42"
assert result.structured_content == {"result": "42"}
async def test_context_injection_async(self):
@ -746,7 +757,8 @@ class TestContextHandling:
async with context:
result = await manager.call_tool("async_tool", {"x": 42})
assert result.content[0].text == "42" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "42"
assert result.structured_content == {"result": "42"}
async def test_context_optional(self):
@ -765,7 +777,8 @@ class TestContextHandling:
async with context:
result = await manager.call_tool("tool_with_context", {"x": 42})
assert result.content[0].text == "42" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "42"
assert result.structured_content == {"result": 42}
def test_parameterized_context_parameter_detection(self):
@ -882,7 +895,8 @@ class TestCustomToolNames:
# Tool should be callable by its custom name
result = await manager.call_tool("custom_multiply", {"a": 5, "b": 3})
assert result.content[0].text == "15" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "15"
assert result.structured_content == {"result": 15}
# Original name should not be registered
@ -1037,7 +1051,8 @@ class TestMountedComponentsRaiseOnLoadError:
# Create a failing mounted server by corrupting it
parent_mcp.mount(child_mcp, namespace="child")
# Corrupt the parent's providers to make it fail during loading
parent_mcp._providers.append("invalid") # type: ignore
assert isinstance(parent_mcp._providers, list)
parent_mcp._providers.append("invalid") # type: ignore[arg-type]
# Should not raise, just warn; use server middleware path now
tools = await parent_mcp._list_tools_middleware()
@ -1051,7 +1066,8 @@ class TestMountedComponentsRaiseOnLoadError:
# Create a failing mounted server
parent_mcp.mount(child_mcp, namespace="child")
# Corrupt the parent's providers to make it fail during loading
parent_mcp._providers.append("invalid") # type: ignore
assert isinstance(parent_mcp._providers, list)
parent_mcp._providers.append("invalid") # type: ignore[arg-type]
# Use temporary settings context manager
with temporary_settings(mounted_components_raise_on_load_error=True):

View file

@ -129,7 +129,8 @@ async def test_hidden_arg_without_default_uses_parent_default(add_tool):
assert sorted(new_tool.parameters["properties"]) == ["old_x"]
# Should pass old_x=3 and let parent use its default old_y=10
result = await new_tool.run(arguments={"old_x": 3})
assert result.content[0].text == "13" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "13"
assert result.structured_content == {"result": 13}
@ -155,7 +156,8 @@ async def test_mixed_hidden_args_with_custom_function(add_tool):
assert sorted(new_tool.parameters["properties"]) == ["visible_x"]
# Should pass visible_x=7 as old_x=7 and old_y=25 to parent
result = await new_tool.run(arguments={"visible_x": 7})
assert result.content[0].text == "32" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "32"
assert result.structured_content == {"result": 32}
@ -238,7 +240,8 @@ async def test_forward_with_argument_mapping(add_tool):
)
result = await new_tool.run(arguments={"new_x": 2, "new_y": 3})
assert result.content[0].text == "5" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "5"
assert result.structured_content == {"result": 5}
@ -279,18 +282,21 @@ async def test_forward_raw_without_argument_mapping(add_tool):
)
result = await new_tool.run(arguments={"new_x": 2, "new_y": 3})
assert result.content[0].text == "5" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "5"
assert result.structured_content == {"result": 5}
async def test_custom_fn_with_kwargs_and_no_transform_args(add_tool):
async def custom_fn(extra: int, **kwargs) -> int:
sum = await forward(**kwargs)
return int(sum.content[0].text) + extra # type: ignore[attr-defined]
assert isinstance(sum.content[0], TextContent)
return int(sum.content[0].text) + extra
new_tool = Tool.from_tool(add_tool, transform_fn=custom_fn)
result = await new_tool.run(arguments={"extra": 1, "old_x": 2, "old_y": 3})
assert result.content[0].text == "6" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "6"
assert result.structured_content == {"result": 6}
assert new_tool.parameters["required"] == IsList(
"extra", "old_x", check_order=False
@ -308,7 +314,8 @@ async def test_fn_with_kwargs_passes_through_original_args(add_tool):
new_tool = Tool.from_tool(add_tool, transform_fn=custom_fn)
result = await new_tool.run(arguments={"new_y": 2, "old_y": 3})
assert result.content[0].text == "5" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "5"
assert result.structured_content == {"result": 5}
@ -327,7 +334,8 @@ async def test_fn_with_kwargs_receives_transformed_arg_names(add_tool):
transform_args={"old_x": ArgTransform(name="new_x")},
)
result = await new_tool.run(arguments={"new_x": 2, "old_y": 3})
assert result.content[0].text == "5" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "5"
assert result.structured_content == {"result": 5}
@ -350,7 +358,8 @@ async def test_fn_with_kwargs_handles_partial_explicit_args(add_tool):
result = await new_tool.run(
arguments={"new_x": 3, "old_y": 7, "some_other_param": "test"}
)
assert result.content[0].text == "10" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "10"
assert result.structured_content == {"result": 10}
@ -369,7 +378,8 @@ async def test_fn_with_kwargs_mixed_mapped_and_unmapped_args(add_tool):
transform_args={"old_x": ArgTransform(name="new_x")},
) # only map 'a'
result = await new_tool.run(arguments={"new_x": 1, "old_y": 5})
assert result.content[0].text == "6" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "6"
assert result.structured_content == {"result": 6}
@ -393,7 +403,8 @@ async def test_fn_with_kwargs_dropped_args_not_in_kwargs(add_tool):
) # drop 'old_y'
result = await new_tool.run(arguments={"new_x": 8})
# 8 + 10 (default value of b in parent)
assert result.content[0].text == "18" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "18"
async def test_forward_outside_context_raises_error():
@ -531,18 +542,21 @@ async def test_tool_transform_chaining(add_tool):
tool2 = Tool.from_tool(tool1, transform_args={"x": ArgTransform(name="final_x")})
result = await tool2.run(arguments={"final_x": 5})
assert result.content[0].text == "15" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "15"
# Transform tool1 with custom function that handles all parameters
async def custom(final_x: int, **kwargs) -> str:
result = await forward(final_x=final_x, **kwargs)
return f"custom {result.content[0].text}" # Extract text from content # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
return f"custom {result.content[0].text}" # Extract text from content
tool3 = Tool.from_tool(
tool1, transform_fn=custom, transform_args={"x": ArgTransform(name="final_x")}
)
result = await tool3.run(arguments={"final_x": 3, "old_y": 5})
assert result.content[0].text == "custom 8" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "custom 8"
class MyModel(BaseModel):
@ -670,7 +684,8 @@ async def test_arg_transform_precedence_over_function_with_kwargs():
# Function signature has different types/defaults than ArgTransform
async def custom_fn(x: str = "function_default", **kwargs) -> str:
result = await forward(x=x, **kwargs)
return f"custom: {result.content[0].text}" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
return f"custom: {result.content[0].text}"
tool = Tool.from_tool(
base,
@ -697,7 +712,8 @@ async def test_arg_transform_precedence_over_function_with_kwargs():
# Test it works at runtime
result = await tool.run(arguments={"y": "test"})
# Should use ArgTransform default of 42
assert "42: test" in result.content[0].text # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "42: test" in result.content[0].text
def test_arg_transform_combined_attributes():
@ -742,7 +758,8 @@ async def test_arg_transform_type_precedence_runtime():
# Convert string back to int for the original function
result = await forward_raw(x=int(x), y=y)
# Extract the text from the result
result_text = result.content[0].text # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
result_text = result.content[0].text
return f"String input '{x}' converted to result: {result_text}"
tool = Tool.from_tool(
@ -754,8 +771,9 @@ async def test_arg_transform_type_precedence_runtime():
# Test it works with string input
result = await tool.run(arguments={"x": "5", "y": 3})
assert "String input '5'" in result.content[0].text # type: ignore[attr-defined]
assert "result: 8" in result.content[0].text # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert "String input '5'" in result.content[0].text
assert "result: 8" in result.content[0].text
class TestProxy:
@ -790,7 +808,8 @@ class TestProxy:
async with Client(proxy_server) as client:
# The tool should be registered with its transformed name
result = await client.call_tool("add_transformed", {"new_x": 1, "old_y": 2})
assert result.content[0].text == "3" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "3"
async def test_arg_transform_default_factory():
@ -813,7 +832,8 @@ async def test_arg_transform_default_factory():
# Should work without providing timestamp (gets value from factory)
result = await new_tool.run(arguments={"x": 42})
assert result.content[0].text == "42_12345.0" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "42_12345.0"
async def test_arg_transform_default_factory_called_each_time():
@ -841,11 +861,13 @@ async def test_arg_transform_default_factory_called_each_time():
# First call
result1 = await new_tool.run(arguments={"x": 1})
assert result1.content[0].text == "1_1" # type: ignore[attr-defined]
assert isinstance(result1.content[0], TextContent)
assert result1.content[0].text == "1_1"
# Second call should get a different value
result2 = await new_tool.run(arguments={"x": 2})
assert result2.content[0].text == "2_2" # type: ignore[attr-defined]
assert isinstance(result2.content[0], TextContent)
assert result2.content[0].text == "2_2"
async def test_arg_transform_hidden_with_default_factory():
@ -870,7 +892,8 @@ async def test_arg_transform_hidden_with_default_factory():
# Should pass hidden request_id with factory value
result = await new_tool.run(arguments={"x": 42})
assert result.content[0].text == "42_req_123" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "42_req_123"
async def test_arg_transform_default_and_factory_raises_error():
@ -907,7 +930,8 @@ async def test_arg_transform_required_true():
# Should work when parameter is provided
result = await new_tool.run(arguments={"optional_param": 100})
assert result.content[0].text == "value: 100" # type: ignore
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "value: 100"
# Should fail when parameter is not provided
with pytest.raises(TypeError, match="Missing required argument"):
@ -925,9 +949,10 @@ async def test_arg_transform_required_false():
ValueError,
match="Cannot specify 'required=False'. Set a default value instead.",
):
# Intentionally passing invalid argument to test error handling
Tool.from_tool(
base_tool,
transform_args={"required_param": ArgTransform(required=False, default=99)}, # type: ignore
transform_args={"required_param": ArgTransform(required=False, default=99)}, # type: ignore[arg-type]
)
@ -954,7 +979,8 @@ async def test_arg_transform_required_with_rename():
# Should work with new name
result = await new_tool.run(arguments={"new_param": 200})
assert result.content[0].text == "value: 200" # type: ignore
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "value: 200"
async def test_arg_transform_required_true_with_default_raises_error():
@ -996,7 +1022,8 @@ async def test_arg_transform_required_no_change():
# Should work as expected
result = await new_tool.run(arguments={"req": 1})
assert result.content[0].text == "values: 1, 42" # type: ignore
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "values: 1, 42"
async def test_arg_transform_hide_and_required_raises_error():
@ -1033,7 +1060,8 @@ class TestEnableDisable:
assert {tool.name for tool in tools} == {"new_add"}
result = await client.call_tool("new_add", {"x": 1, "y": 2})
assert result.content[0].text == "3" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "3"
with pytest.raises(ToolError):
await client.call_tool("add", {"x": 1, "y": 2})
@ -1109,7 +1137,8 @@ class TestTransformToolOutputSchema:
result = await new_tool.run({"x": 5})
# Even with output_schema=None, structured content should be generated via fallback logic
assert result.structured_content == {"result": "Result: 5"}
assert result.content[0].text == "Result: 5" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Result: 5"
def test_transform_with_explicit_output_schema_dict(self, base_string_tool):
"""Test that explicit output schema overrides parent."""
@ -1130,14 +1159,16 @@ class TestTransformToolOutputSchema:
result = await new_tool.run({"x": 10})
# Non-object explicit schemas disable structured content
assert result.structured_content is None
assert result.content[0].text == "Result: 10" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Result: 10"
def test_transform_with_custom_function_inferred_schema(self, base_dict_tool):
"""Test that custom function's output schema is inferred."""
async def custom_fn(x: int) -> str:
result = await forward(x=x)
return f"Custom: {result.content[0].text}" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
return f"Custom: {result.content[0].text}"
new_tool = Tool.from_tool(base_dict_tool, transform_fn=custom_fn)
@ -1155,7 +1186,8 @@ class TestTransformToolOutputSchema:
async def custom_fn(x: int) -> str:
result = await forward(x=x)
return f"Custom: {result.content[0].text}" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
return f"Custom: {result.content[0].text}"
new_tool = Tool.from_tool(base_dict_tool, transform_fn=custom_fn)
@ -1202,7 +1234,8 @@ class TestTransformToolOutputSchema:
# Object types should not be wrapped
expected_schema = TypeAdapter(dict[str, int]).json_schema()
assert new_tool.output_schema == expected_schema
assert "x-fastmcp-wrap-result" not in new_tool.output_schema # type: ignore[attr-defined]
assert isinstance(new_tool.output_schema, dict)
assert "x-fastmcp-wrap-result" not in new_tool.output_schema
result = await new_tool.run({"x": 4})
# Direct value, not wrapped
@ -1215,7 +1248,8 @@ class TestTransformToolOutputSchema:
result = await new_tool.run({"x": 7})
# Should wrap because parent schema has wrap marker
assert result.structured_content == {"result": "Result: 7"}
assert "x-fastmcp-wrap-result" in new_tool.output_schema # type: ignore[attr-defined]
assert isinstance(new_tool.output_schema, dict)
assert "x-fastmcp-wrap-result" in new_tool.output_schema
def test_transform_chained_output_schema_inheritance(self, base_string_tool):
"""Test output schema inheritance through multiple transformations."""
@ -1260,14 +1294,16 @@ class TestTransformToolOutputSchema:
# Test ToolResult return
result2 = await new_tool.run({"x": 2})
assert result2.structured_content == {"custom_value": 2}
assert result2.content[0].text == "Custom: 2" # type: ignore[attr-defined]
assert isinstance(result2.content[0], TextContent)
assert result2.content[0].text == "Custom: 2"
def test_transform_output_schema_with_arg_transforms(self, base_string_tool):
"""Test that output schema works correctly with argument transformations."""
async def custom_fn(new_x: int) -> dict[str, str]:
result = await forward(new_x=new_x)
return {"transformed": result.content[0].text} # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
return {"transformed": result.content[0].text}
new_tool = Tool.from_tool(
base_string_tool,
@ -1299,7 +1335,9 @@ class TestTransformToolOutputSchema:
assert result_explicit_none.structured_content == {
"result": "Result: 5"
} # Generated via fallback logic
assert result_default.content[0].text == result_explicit_none.content[0].text # type: ignore[attr-defined]
assert isinstance(result_default.content[0], TextContent)
assert isinstance(result_explicit_none.content[0], TextContent)
assert result_default.content[0].text == result_explicit_none.content[0].text
async def test_transform_output_schema_with_tool_result_return(
self, base_string_tool
@ -1320,7 +1358,8 @@ class TestTransformToolOutputSchema:
result = await new_tool.run({"x": 6})
# Should use ToolResult content directly
assert result.content[0].text == "Direct: 6" # type: ignore[attr-defined]
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Direct: 6"
assert result.structured_content == {"direct_value": 6, "doubled": 12}

View file

@ -1569,7 +1569,6 @@ class TestFieldsWithDefaults:
generated_type = json_schema_to_type(schema)
validator = TypeAdapter(generated_type)
result = validator.validate_python({})
assert result.flag is False # type: ignore[attr-defined]
def test_field_with_default_accepts_explicit_value(self):
@ -1582,5 +1581,4 @@ class TestFieldsWithDefaults:
generated_type = json_schema_to_type(schema)
validator = TypeAdapter(generated_type)
result = validator.validate_python({"flag": True})
assert result.flag is True # type: ignore[attr-defined]