ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-22 05:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #646 from jlowin/fix-typing

Browse source 
Update test typing
This commit is contained in:
Jeremiah Lowin 2025-05-31 20:49:16 -04:00 • committed by GitHub
commit 92872a4060
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 222 additions and 521 deletions
Download patch file Download diff file Expand all files Collapse all files

22
tests/client/test_client.py
View file

@ -16,7 +16,6 @@ from fastmcp.client.transports import (
infer_transport,
)
from fastmcp.exceptions import ResourceError, ToolError
from fastmcp.prompts.prompt import TextContent
from fastmcp.server.server import FastMCP
@ -201,8 +200,7 @@ async def test_get_prompt(fastmcp_server):
result = await client.get_prompt("welcome", {"name": "Developer"})
# The result should contain our welcome message
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Welcome to FastMCP, Developer!"
assert result.messages[0].content.text == "Welcome to FastMCP, Developer!" # type: ignore[attr-defined]
assert result.description == "Example greeting prompt."
@ -214,8 +212,7 @@ 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 isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Welcome to FastMCP, Developer!"
assert result.messages[0].content.text == "Welcome to FastMCP, Developer!" # type: ignore[attr-defined]
assert result.description == "Example greeting prompt."
@ -522,9 +519,8 @@ class TestErrorHandling:
async with client:
result = await client.call_tool_mcp("error_tool", {})
assert result.isError
assert isinstance(result.content[0], TextContent)
assert "test error" in result.content[0].text
assert "abc" in result.content[0].text
assert "test error" in result.content[0].text # type: ignore[attr-defined]
assert "abc" in result.content[0].text # type: ignore[attr-defined]
async def test_general_tool_exceptions_are_masked_when_enabled(self):
mcp = FastMCP("TestServer", mask_error_details=True)
@ -538,9 +534,8 @@ class TestErrorHandling:
async with client:
result = await client.call_tool_mcp("error_tool", {})
assert result.isError
assert isinstance(result.content[0], TextContent)
assert "test error" not in result.content[0].text
assert "abc" not in result.content[0].text
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]
async def test_specific_tool_errors_are_sent_to_client(self):
mcp = FastMCP("TestServer")
@ -554,9 +549,8 @@ class TestErrorHandling:
async with client:
result = await client.call_tool_mcp("custom_error_tool", {})
assert result.isError
assert isinstance(result.content[0], TextContent)
assert "test error" in result.content[0].text
assert "abc" in result.content[0].text
assert "test error" in result.content[0].text # type: ignore[attr-defined]
assert "abc" in result.content[0].text # type: ignore[attr-defined]
async def test_general_resource_exceptions_are_not_masked_by_default(self):
mcp = FastMCP("TestServer")

28
tests/client/test_openapi.py
View file

@ -5,7 +5,6 @@ from collections.abc import Generator
import pytest
import uvicorn
from fastapi import FastAPI, Request
from mcp.types import TextContent, TextResourceContents
from fastmcp import Client, FastMCP
from fastmcp.client.transports import SSETransport, StreamableHttpTransport
@ -111,8 +110,7 @@ class TestClientHeaders:
transport=SSETransport(sse_server, headers={"X-TEST": "test-123"})
) as client:
result = await client.read_resource("resource://get_headers_headers_get")
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert headers["x-test"] == "test-123"
async def test_client_headers_shttp_resource(self, shttp_server: str):
@ -122,8 +120,7 @@ class TestClientHeaders:
)
) as client:
result = await client.read_resource("resource://get_headers_headers_get")
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert headers["x-test"] == "test-123"
async def test_client_headers_sse_resource_template(self, sse_server: str):
@ -133,8 +130,7 @@ class TestClientHeaders:
result = await client.read_resource(
"resource://get_header_by_name_headers/x-test"
)
assert isinstance(result[0], TextResourceContents)
header = json.loads(result[0].text)
header = json.loads(result[0].text) # type: ignore[attr-defined]
assert header == "test-123"
async def test_client_headers_shttp_resource_template(self, shttp_server: str):
@ -146,8 +142,7 @@ class TestClientHeaders:
result = await client.read_resource(
"resource://get_header_by_name_headers/x-test"
)
assert isinstance(result[0], TextResourceContents)
header = json.loads(result[0].text)
header = json.loads(result[0].text) # type: ignore[attr-defined]
assert header == "test-123"
async def test_client_headers_sse_tool(self, sse_server: str):
@ -155,8 +150,7 @@ class TestClientHeaders:
transport=SSETransport(sse_server, headers={"X-TEST": "test-123"})
) as client:
result = await client.call_tool("post_headers_headers_post")
assert isinstance(result[0], TextContent)
headers = json.loads(result[0].text)
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert headers["x-test"] == "test-123"
async def test_client_headers_shttp_tool(self, shttp_server: str):
@ -166,8 +160,7 @@ class TestClientHeaders:
)
) as client:
result = await client.call_tool("post_headers_headers_post")
assert isinstance(result[0], TextContent)
headers = json.loads(result[0].text)
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert headers["x-test"] == "test-123"
async def test_client_overrides_server_headers(self, shttp_server: str):
@ -177,8 +170,7 @@ class TestClientHeaders:
)
) as client:
result = await client.read_resource("resource://get_headers_headers_get")
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert headers["x-server-header"] == "test-client"
async def test_client_with_excluded_header_is_ignored(self, sse_server: str):
@ -193,8 +185,7 @@ class TestClientHeaders:
)
) as client:
result = await client.read_resource("resource://get_headers_headers_get")
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert headers["not-host"] == "1.2.3.4"
assert headers["host"] == "fastapi"
@ -204,6 +195,5 @@ class TestClientHeaders:
"""
async with Client(transport=StreamableHttpTransport(proxy_server)) as client:
result = await client.read_resource("resource://get_headers_headers_get")
assert isinstance(result[0], TextResourceContents)
headers = json.loads(result[0].text)
headers = json.loads(result[0].text) # type: ignore[attr-defined]
assert headers["x-server-header"] == "test-abc"

4
tests/client/test_roots.py
View file

@ -1,7 +1,6 @@
import json
import pytest
from mcp.types import TextContent
from fastmcp import Client, Context, FastMCP
@ -41,8 +40,7 @@ class TestClientRoots:
async def test_valid_roots(self, fastmcp_server: FastMCP, roots: list[str]):
async with Client(fastmcp_server, roots=roots) as client:
result = await client.call_tool("list_roots", {})
assert isinstance(result[0], TextContent)
assert json.loads(result[0].text) == [
assert json.loads(result[0].text) == [ # type: ignore[attr-defined]
"file://x/y/z",
"file://x/y/z",
]

4
tests/client/test_sse.py
View file

@ -6,7 +6,6 @@ from collections.abc import Generator
import pytest
import uvicorn
from mcp import McpError
from mcp.types import TextResourceContents
from starlette.applications import Starlette
from starlette.routing import Mount
@ -96,8 +95,7 @@ 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")
assert isinstance(raw_result[0], TextResourceContents)
json_result = json.loads(raw_result[0].text)
json_result = json.loads(raw_result[0].text) # type: ignore[attr-defined]
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"

28
tests/client/test_stdio.py
View file

@ -1,7 +1,6 @@
import inspect
import pytest
from mcp.types import TextContent
from fastmcp import Client
from fastmcp.client.transports import PythonStdioTransport, StdioTransport
@ -49,13 +48,11 @@ class TestKeepAlive:
async with client:
result1 = await client.call_tool("pid")
assert isinstance(result1[0], TextContent)
pid1 = int(result1[0].text)
pid1 = int(result1[0].text) # type: ignore[attr-defined]
async with client:
result2 = await client.call_tool("pid")
assert isinstance(result2[0], TextContent)
pid2 = int(result2[0].text)
pid2 = int(result2[0].text) # type: ignore[attr-defined]
assert pid1 == pid2
@ -69,13 +66,11 @@ class TestKeepAlive:
async with client:
result1 = await client.call_tool("pid")
assert isinstance(result1[0], TextContent)
pid1 = int(result1[0].text)
pid1 = int(result1[0].text) # type: ignore[attr-defined]
async with client:
result2 = await client.call_tool("pid")
assert isinstance(result2[0], TextContent)
pid2 = int(result2[0].text)
pid2 = int(result2[0].text) # type: ignore[attr-defined]
assert pid1 != pid2
@ -85,15 +80,13 @@ class TestKeepAlive:
async with client:
result1 = await client.call_tool("pid")
assert isinstance(result1[0], TextContent)
pid1 = int(result1[0].text)
pid1 = int(result1[0].text) # type: ignore[attr-defined]
await client.close()
async with client:
result2 = await client.call_tool("pid")
assert isinstance(result2[0], TextContent)
pid2 = int(result2[0].text)
pid2 = int(result2[0].text) # type: ignore[attr-defined]
assert pid1 != pid2
@ -103,17 +96,14 @@ class TestKeepAlive:
async with client:
result1 = await client.call_tool("pid")
assert isinstance(result1[0], TextContent)
pid1 = int(result1[0].text)
pid1 = int(result1[0].text) # type: ignore[attr-defined]
async with client:
result2 = await client.call_tool("pid")
assert isinstance(result2[0], TextContent)
pid2 = int(result2[0].text)
pid2 = int(result2[0].text) # type: ignore[attr-defined]
result3 = await client.call_tool("pid")
assert isinstance(result3[0], TextContent)
pid3 = int(result3[0].text)
pid3 = int(result3[0].text) # type: ignore[attr-defined]
assert pid1 == pid2 == pid3

4
tests/client/test_streamable_http.py
View file

@ -6,7 +6,6 @@ from collections.abc import Generator
import pytest
import uvicorn
from mcp import McpError
from mcp.types import TextResourceContents
from starlette.applications import Starlette
from starlette.routing import Mount
@ -106,8 +105,7 @@ async def test_http_headers(streamable_http_server: str):
)
) as client:
raw_result = await client.read_resource("request://headers")
assert isinstance(raw_result[0], TextResourceContents)
json_result = json.loads(raw_result[0].text)
json_result = json.loads(raw_result[0].text) # type: ignore[attr-defined]
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"

6
tests/prompts/test_prompt_manager.py
View file

@ -393,8 +393,7 @@ class TestContextHandling:
messages = await prompt.render(arguments={"x": 42})
assert len(messages) == 1
assert isinstance(messages[0].content, TextContent)
assert messages[0].content.text == "42"
assert messages[0].content.text == "42" # type: ignore[attr-defined]
async def test_context_optional(self):
"""Test that context is optional when rendering prompts."""
@ -416,8 +415,7 @@ class TestContextHandling:
)
assert len(messages) == 1
assert isinstance(messages[0].content, TextContent)
assert messages[0].content.text == "42"
assert messages[0].content.text == "42" # type: ignore[attr-defined]
async def test_annotated_context_parameter_detection(self):
"""Test that annotated context parameters are properly detected in

1
tests/resources/test_file_resources.py
View file

@ -74,7 +74,6 @@ class TestFileResource:
is_binary=True,
)
content = await resource.read()
assert isinstance(content, bytes)
assert content == b"test content"
def test_relative_path_error(self):

19
tests/server/http/test_http_dependencies.py
View file

@ -4,7 +4,6 @@ from collections.abc import Generator
import pytest
import uvicorn
from mcp.types import TextContent, TextResourceContents
from fastmcp.client import Client
from fastmcp.client.transports import SSETransport, StreamableHttpTransport
@ -99,8 +98,7 @@ async def test_http_headers_resource_shttp(shttp_server: str):
)
) as client:
raw_result = await client.read_resource("request://headers")
assert isinstance(raw_result[0], TextResourceContents)
json_result = json.loads(raw_result[0].text)
json_result = json.loads(raw_result[0].text) # type: ignore[attr-defined]
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"
@ -111,8 +109,7 @@ 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")
assert isinstance(raw_result[0], TextResourceContents)
json_result = json.loads(raw_result[0].text)
json_result = json.loads(raw_result[0].text) # type: ignore[attr-defined]
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"
@ -125,8 +122,7 @@ async def test_http_headers_tool_shttp(shttp_server: str):
)
) as client:
result = await client.call_tool("get_headers_tool")
assert isinstance(result[0], TextContent)
json_result = json.loads(result[0].text)
json_result = json.loads(result[0].text) # type: ignore[attr-defined]
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"
@ -136,8 +132,7 @@ async def test_http_headers_tool_sse(sse_server: str):
transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
) as client:
result = await client.call_tool("get_headers_tool")
assert isinstance(result[0], TextContent)
json_result = json.loads(result[0].text)
json_result = json.loads(result[0].text) # type: ignore[attr-defined]
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"
@ -150,8 +145,7 @@ async def test_http_headers_prompt_shttp(shttp_server: str):
)
) as client:
result = await client.get_prompt("get_headers_prompt")
assert isinstance(result.messages[0].content, TextContent)
json_result = json.loads(result.messages[0].content.text)
json_result = json.loads(result.messages[0].content.text) # type: ignore[attr-defined]
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"
@ -162,7 +156,6 @@ 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")
assert isinstance(result.messages[0].content, TextContent)
json_result = json.loads(result.messages[0].content.text)
json_result = json.loads(result.messages[0].content.text) # type: ignore[attr-defined]
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"

54
tests/server/openapi/test_openapi.py
View file

@ -9,7 +9,7 @@ from dirty_equals import IsStr
from fastapi import FastAPI, HTTPException, Response
from fastapi.responses import PlainTextResponse
from httpx import ASGITransport, AsyncClient
from mcp.types import BlobResourceContents, TextContent, TextResourceContents
from mcp.types import BlobResourceContents
from pydantic import BaseModel, TypeAdapter
from pydantic.networks import AnyUrl
@ -234,11 +234,7 @@ class TestTools:
"create_user_users_post", {"name": "David", "active": False}
)
# Convert TextContent to dict for comparison
assert isinstance(tool_response, list) and len(tool_response) == 1
assert isinstance(tool_response[0], TextContent)
response_data = json.loads(tool_response[0].text)
response_data = json.loads(tool_response[0].text) # type: ignore[attr-defined]
expected_user = User(id=4, name="David", active=False).model_dump()
assert response_data == expected_user
@ -249,8 +245,7 @@ class TestTools:
# Check that the user was created via MCP
async with Client(fastmcp_openapi_server) as client:
user_response = await client.read_resource("resource://get_user_users/4")
assert isinstance(user_response[0], TextResourceContents)
response_text = user_response[0].text
response_text = user_response[0].text # type: ignore[attr-defined]
user = json.loads(response_text)
assert user == expected_user
@ -266,11 +261,7 @@ class TestTools:
{"user_id": 1, "name": "XYZ"},
)
# Convert TextContent to dict for comparison
assert isinstance(tool_response, list) and len(tool_response) == 1
assert isinstance(tool_response[0], TextContent)
response_data = json.loads(tool_response[0].text)
response_data = json.loads(tool_response[0].text) # type: ignore[attr-defined]
expected_data = dict(id=1, name="XYZ", active=True)
assert response_data == expected_data
@ -281,8 +272,7 @@ class TestTools:
# Check that the user was updated via MCP
async with Client(fastmcp_openapi_server) as client:
user_response = await client.read_resource("resource://get_user_users/1")
assert isinstance(user_response[0], TextResourceContents)
response_text = user_response[0].text
response_text = user_response[0].text # type: ignore[attr-defined]
user = json.loads(response_text)
assert user == expected_data
@ -305,9 +295,7 @@ class TestTools:
)
async with Client(mcp_server) as client:
tool_response = await client.call_tool("get_users_users_get", {})
assert isinstance(tool_response, list)
assert isinstance(tool_response[0], TextContent)
assert json.loads(tool_response[0].text) == [
assert json.loads(tool_response[0].text) == [ # type: ignore[attr-defined]
user.model_dump()
for user in sorted(users_db.values(), key=lambda x: x.id)
]
@ -341,8 +329,7 @@ class TestResources:
resource_response = await client.read_resource(
"resource://get_users_users_get"
)
assert isinstance(resource_response[0], TextResourceContents)
response_text = resource_response[0].text
response_text = resource_response[0].text # type: ignore[attr-defined]
resource = json.loads(response_text)
assert resource == json_users
response = await api_client.get("/users")
@ -369,8 +356,7 @@ class TestResources:
"""Test reading a resource that returns a string."""
async with Client(fastmcp_openapi_server) as client:
resource_response = await client.read_resource("resource://ping_ping_get")
assert isinstance(resource_response[0], TextResourceContents)
assert resource_response[0].text == "pong"
assert resource_response[0].text == "pong" # type: ignore[attr-defined]
class TestResourceTemplates:
@ -407,8 +393,7 @@ class TestResourceTemplates:
resource_response = await client.read_resource(
f"resource://get_user_users/{user_id}"
)
assert isinstance(resource_response[0], TextResourceContents)
response_text = resource_response[0].text
response_text = resource_response[0].text # type: ignore[attr-defined]
resource = json.loads(response_text)
assert resource == users_db[user_id].model_dump()
@ -430,8 +415,7 @@ class TestResourceTemplates:
resource_response = await client.read_resource(
f"resource://get_user_active_state_users/{is_active}/{user_id}"
)
assert isinstance(resource_response[0], TextResourceContents)
response_text = resource_response[0].text
response_text = resource_response[0].text # type: ignore[attr-defined]
resource = json.loads(response_text)
assert resource == users_db[user_id].model_dump()
@ -681,8 +665,7 @@ class TestOpenAPI30Compatibility:
"""Test reading a resource from an OpenAPI 3.0 server."""
async with Client(openapi_30_server) as client:
resource_response = await client.read_resource("resource://listProducts")
assert isinstance(resource_response[0], TextResourceContents)
response_text = resource_response[0].text
response_text = resource_response[0].text # type: ignore[attr-defined]
content = json.loads(response_text)
assert len(content) == 2
assert content[0]["name"] == "Product 1"
@ -692,8 +675,7 @@ class TestOpenAPI30Compatibility:
"""Test reading a resource from template from an OpenAPI 3.0 server."""
async with Client(openapi_30_server) as client:
resource_response = await client.read_resource("resource://getProduct/p1")
assert isinstance(resource_response[0], TextResourceContents)
response_text = resource_response[0].text
response_text = resource_response[0].text # type: ignore[attr-defined]
content = json.loads(response_text)
assert content["id"] == "p1"
assert content["name"] == "Product 1"
@ -707,8 +689,7 @@ class TestOpenAPI30Compatibility:
)
# Result should be a text content
assert len(result) == 1
assert isinstance(result[0], TextContent)
product = json.loads(result[0].text)
product = json.loads(result[0].text) # type: ignore[attr-defined]
assert product["id"] == "p3"
assert product["name"] == "New Product"
assert product["price"] == 39.99
@ -857,8 +838,7 @@ class TestOpenAPI31Compatibility:
"""Test reading a resource from an OpenAPI 3.1 server."""
async with Client(openapi_31_server) as client:
resource_response = await client.read_resource("resource://listOrders")
assert isinstance(resource_response[0], TextResourceContents)
response_text = resource_response[0].text
response_text = resource_response[0].text # type: ignore[attr-defined]
content = json.loads(response_text)
assert len(content) == 2
assert content[0]["customer"] == "Alice"
@ -868,8 +848,7 @@ class TestOpenAPI31Compatibility:
"""Test reading a resource from template from an OpenAPI 3.1 server."""
async with Client(openapi_31_server) as client:
resource_response = await client.read_resource("resource://getOrder/o1")
assert isinstance(resource_response[0], TextResourceContents)
response_text = resource_response[0].text
response_text = resource_response[0].text # type: ignore[attr-defined]
content = json.loads(response_text)
assert content["id"] == "o1"
assert content["customer"] == "Alice"
@ -883,8 +862,7 @@ class TestOpenAPI31Compatibility:
)
# Result should be a text content
assert len(result) == 1
assert isinstance(result[0], TextContent)
order = json.loads(result[0].text)
order = json.loads(result[0].text) # type: ignore[attr-dict]
assert order["id"] == "o3"
assert order["customer"] == "Charlie"
assert order["items"] == ["item4", "item5"]

20
tests/server/test_import_server.py
View file

@ -1,8 +1,6 @@
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
@ -223,8 +221,7 @@ async def test_call_imported_custom_named_tool():
async with Client(main_app) as client:
result = await client.call_tool("api_get_data", {"query": "test"})
assert isinstance(result[0], TextContent)
assert result[0].text == "Data for query: test"
assert result[0].text == "Data for query: test" # type: ignore[attr-defined]
async def test_first_level_importing_with_custom_name():
@ -278,8 +275,7 @@ async def test_call_nested_imported_tool():
result = await main_app._tool_manager.call_tool(
"service_provider_compute", {"input": 21}
)
assert isinstance(result[0], TextContent)
assert result[0].text == "42"
assert result[0].text == "42" # type: ignore[attr-defined]
async def test_import_with_proxy_tools():
@ -302,8 +298,7 @@ async def test_import_with_proxy_tools():
await main_app.import_server("api", proxy_app)
result = await main_app._mcp_call_tool("api_get_data", {"query": "test"})
assert isinstance(result[0], TextContent)
assert result[0].text == "Data for query: test"
assert result[0].text == "Data for query: test" # type: ignore[attr-defined]
async def test_import_with_proxy_prompts():
@ -326,8 +321,7 @@ async def test_import_with_proxy_prompts():
await main_app.import_server("api", proxy_app)
result = await main_app._mcp_get_prompt("api_greeting", {"name": "World"})
assert isinstance(result.messages[0].content, TextContent)
assert result.messages[0].content.text == "Hello, World from API!"
assert result.messages[0].content.text == "Hello, World from API!" # type: ignore[attr-defined]
assert result.description == "Example greeting prompt."
@ -356,8 +350,7 @@ 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")
assert isinstance(result[0], TextResourceContents)
content = json.loads(result[0].text)
content = json.loads(result[0].text) # type: ignore[attr-defined]
assert content["api_key"] == "12345"
assert content["base_url"] == "https://api.example.com"
@ -387,8 +380,7 @@ 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}")
assert isinstance(result[0], TextResourceContents)
content = json.loads(result[0].text)
content = json.loads(result[0].text) # type: ignore[attr-defined]
assert content["name"] == "John Doe"
assert content["email"] == "john@example.com"

38
tests/server/test_mount.py
View file

@ -3,8 +3,6 @@ import sys
from contextlib import asynccontextmanager
import pytest
from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.types import TextContent, TextResourceContents
from fastmcp import FastMCP
from fastmcp.client import Client
@ -36,8 +34,7 @@ class TestBasicMount:
async with Client(main_app) as client:
result = await client.call_tool("sub_sub_tool", {})
assert isinstance(result[0], TextContent)
assert result[0].text == "This is from the sub app"
assert result[0].text == "This is from the sub app" # type: ignore[attr-defined]
async def test_mount_with_custom_separator(self):
"""Test mounting with a custom tool separator (deprecated but still supported)."""
@ -57,8 +54,7 @@ class TestBasicMount:
# Call the tool
result = await main_app._mcp_call_tool("sub_greet", {"name": "World"})
assert isinstance(result[0], TextContent)
assert result[0].text == "Hello, World!"
assert result[0].text == "Hello, World!" # type: ignore[attr-defined]
async def test_mount_invalid_resource_prefix(self):
main_app = FastMCP("MainApp")
@ -147,12 +143,10 @@ class TestMultipleServerMount:
# Call tools from both mounted servers
result1 = await main_app._mcp_call_tool("weather_get_forecast", {})
assert isinstance(result1[0], TextContent)
assert result1[0].text == "Weather forecast"
assert result1[0].text == "Weather forecast" # type: ignore[attr-defined]
result2 = await main_app._mcp_call_tool("news_get_headlines", {})
assert isinstance(result2[0], TextContent)
assert result2[0].text == "News headlines"
assert result2[0].text == "News headlines" # type: ignore[attr-defined]
async def test_mount_same_prefix(self):
"""Test that mounting with the same prefix replaces the previous mount."""
@ -227,8 +221,7 @@ class TestMultipleServerMount:
# Test calling a tool
result = await client.call_tool("working_working_tool", {})
assert isinstance(result[0], TextContent)
assert result[0].text == "Working tool"
assert result[0].text == "Working tool" # type: ignore[attr-defined]
# Test resources
resources = await client.list_resources()
@ -284,8 +277,7 @@ class TestDynamicChanges:
# Call the dynamically added tool
result = await main_app._mcp_call_tool("sub_dynamic_tool", {})
assert isinstance(result[0], TextContent)
assert result[0].text == "Added after mounting"
assert result[0].text == "Added after mounting" # type: ignore[attr-defined]
async def test_removing_tool_after_mounting(self):
"""Test that tools removed from mounted servers are no longer accessible."""
@ -335,8 +327,7 @@ class TestResourcesAndTemplates:
# Check that resource can be accessed
async with Client(main_app) as client:
result = await client.read_resource("data://data/users")
assert isinstance(result[0], TextResourceContents)
assert json.loads(result[0].text) == ["user1", "user2"]
assert json.loads(result[0].text) == ["user1", "user2"] # type: ignore[attr-defined]
async def test_mount_with_resource_templates(self):
"""Test mounting a server with resource templates."""
@ -357,8 +348,7 @@ class TestResourcesAndTemplates:
# Check template instantiation
async with Client(main_app) as client:
result = await client.read_resource("users://api/123/profile")
assert isinstance(result[0], TextResourceContents)
profile = json.loads(result[0].text)
profile = json.loads(result[0].text) # type: ignore
assert profile["id"] == "123"
assert profile["name"] == "User 123"
@ -382,8 +372,7 @@ class TestResourcesAndTemplates:
# Check access to the resource
async with Client(main_app) as client:
result = await client.read_resource("data://data/config")
assert isinstance(result[0], TextResourceContents)
config = json.loads(result[0].text)
config = json.loads(result[0].text) # type: ignore[attr-defined]
assert config["version"] == "1.0"
@ -461,8 +450,7 @@ class TestProxyServer:
# Call the tool
result = await main_app._mcp_call_tool("proxy_get_data", {"query": "test"})
assert isinstance(result[0], TextContent)
assert result[0].text == "Data for test"
assert result[0].text == "Data for test" # type: ignore[attr-defined]
async def test_dynamically_adding_to_proxied_server(self):
"""Test that changes to the original server are reflected in the mounted proxy."""
@ -489,8 +477,7 @@ class TestProxyServer:
# Call the tool
result = await main_app._mcp_call_tool("proxy_dynamic_data", {})
assert isinstance(result[0], TextContent)
assert result[0].text == "Dynamic data"
assert result[0].text == "Dynamic data" # type: ignore[attr-defined]
async def test_proxy_server_with_resources(self):
"""Test mounting a proxy server with resources."""
@ -512,8 +499,7 @@ class TestProxyServer:
# Resource should be accessible through main app
result = await main_app._mcp_read_resource("config://proxy/settings")
assert isinstance(result[0], ReadResourceContents)
config = json.loads(result[0].content)
config = json.loads(result[0].content) # type: ignore[attr-defined]
assert config["api_key"] == "12345"
async def test_proxy_server_with_prompts(self):

24
tests/server/test_proxy.py
View file

@ -1,7 +1,6 @@
import json
from typing import Any
import mcp.types
import pytest
from anyio import create_task_group
from dirty_equals import Contains
@ -90,16 +89,14 @@ async def test_as_proxy_with_server(fastmcp_server):
"""FastMCP.as_proxy should accept a FastMCP instance."""
proxy = FastMCP.as_proxy(fastmcp_server)
result = await proxy._mcp_call_tool("greet", {"name": "Test"})
assert isinstance(result[0], mcp.types.TextContent)
assert result[0].text == "Hello, Test!"
assert result[0].text == "Hello, Test!" # type: ignore[attr-defined]
async def test_as_proxy_with_transport(fastmcp_server):
"""FastMCP.as_proxy should accept a ClientTransport."""
proxy = FastMCP.as_proxy(FastMCPTransport(fastmcp_server))
result = await proxy._mcp_call_tool("greet", {"name": "Test"})
assert isinstance(result[0], mcp.types.TextContent)
assert result[0].text == "Hello, Test!"
assert result[0].text == "Hello, Test!" # type: ignore[attr-defined]
def test_as_proxy_with_url():
@ -138,9 +135,7 @@ class TestTools:
async def test_call_tool_calls_tool(self, proxy_server):
async with Client(proxy_server) as client:
proxy_result = await client.call_tool("add", {"a": 1, "b": 2})
assert isinstance(proxy_result[0], mcp.types.TextContent)
assert proxy_result[0].text == "3"
assert proxy_result[0].text == "3" # type: ignore[attr-defined]
async def test_error_tool_raises_error(self, proxy_server):
with pytest.raises(ToolError, match=""):
@ -164,8 +159,7 @@ 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 isinstance(result[0], mcp.types.TextResourceContents)
assert result[0].text == "👋"
assert result[0].text == "👋" # type: ignore[attr-defined]
async def test_read_resource_same_as_original(self, fastmcp_server, proxy_server):
async with Client(fastmcp_server) as client:
@ -177,8 +171,7 @@ 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 isinstance(result[0], mcp.types.TextResourceContents)
assert json.loads(result[0].text) == USERS
assert json.loads(result[0].text) == USERS # type: ignore[attr-defined]
async def test_read_resource_returns_none_if_not_found(self, proxy_server):
with pytest.raises(McpError, match="Unknown resource: resource://nonexistent"):
@ -202,8 +195,7 @@ 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 isinstance(result[0], mcp.types.TextResourceContents)
assert json.loads(result[0].text) == USERS[id - 1]
assert json.loads(result[0].text) == USERS[id - 1] # type: ignore[attr-defined]
async def test_read_resource_template_same_as_original(
self, fastmcp_server, proxy_server
@ -239,10 +231,8 @@ class TestPrompts:
async def test_render_prompt_calls_prompt(self, proxy_server):
async with Client(proxy_server) as client:
result = await client.get_prompt("welcome", {"name": "Alice"})
assert isinstance(result.messages[0], mcp.types.PromptMessage)
assert result.messages[0].role == "user"
assert isinstance(result.messages[0].content, mcp.types.TextContent)
assert result.messages[0].content.text == "Welcome to FastMCP, Alice!"
assert result.messages[0].content.text == "Welcome to FastMCP, Alice!" # type: ignore[attr-defined]
async def test_proxy_handles_multiple_concurrent_tasks_correctly(

115
tests/server/test_server.py
View file

@ -2,10 +2,6 @@ from typing import Annotated
import pytest
from mcp import McpError
from mcp.types import (
TextContent,
TextResourceContents,
)
from pydantic import Field
from fastmcp import Client, FastMCP
@ -48,8 +44,7 @@ class TestCreateServer:
result = await client.call_tool("hello_world", {})
assert len(result) == 1
content = result[0]
assert isinstance(content, TextContent)
assert "¡Hola, 世界! 👋" == content.text
assert content.text == "¡Hola, 世界! 👋" # type: ignore[attr-defined]
class TestTools:
@ -114,8 +109,7 @@ class TestToolDecorator:
return x + y
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
assert isinstance(result[0], TextContent)
assert result[0].text == "3"
assert result[0].text == "3" # type: ignore[attr-defined]
async def test_tool_decorator_incorrect_usage(self):
mcp = FastMCP()
@ -134,8 +128,7 @@ class TestToolDecorator:
return x + y
result = await mcp._mcp_call_tool("custom-add", {"x": 1, "y": 2})
assert isinstance(result[0], TextContent)
assert result[0].text == "3"
assert result[0].text == "3" # type: ignore[attr-defined]
async def test_tool_decorator_with_description(self):
mcp = FastMCP()
@ -163,8 +156,7 @@ class TestToolDecorator:
obj = MyClass(10)
mcp.add_tool(obj.add)
result = await mcp._mcp_call_tool("add", {"y": 2})
assert isinstance(result[0], TextContent)
assert result[0].text == "12"
assert result[0].text == "12" # type: ignore[attr-defined]
async def test_tool_decorator_classmethod(self):
mcp = FastMCP()
@ -178,8 +170,7 @@ class TestToolDecorator:
mcp.add_tool(MyClass.add)
result = await mcp._mcp_call_tool("add", {"y": 2})
assert isinstance(result[0], TextContent)
assert result[0].text == "12"
assert result[0].text == "12" # type: ignore[attr-defined]
async def test_tool_decorator_staticmethod(self):
mcp = FastMCP()
@ -191,8 +182,7 @@ class TestToolDecorator:
return x + y
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
assert isinstance(result[0], TextContent)
assert result[0].text == "3"
assert result[0].text == "3" # type: ignore[attr-defined]
async def test_tool_decorator_async_function(self):
mcp = FastMCP()
@ -202,8 +192,7 @@ class TestToolDecorator:
return x + y
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
assert isinstance(result[0], TextContent)
assert result[0].text == "3"
assert result[0].text == "3" # type: ignore[attr-defined]
async def test_tool_decorator_classmethod_async_function(self):
mcp = FastMCP()
@ -217,8 +206,7 @@ class TestToolDecorator:
mcp.add_tool(MyClass.add)
result = await mcp._mcp_call_tool("add", {"y": 2})
assert isinstance(result[0], TextContent)
assert result[0].text == "12"
assert result[0].text == "12" # type: ignore[attr-defined]
async def test_tool_decorator_staticmethod_async_function(self):
mcp = FastMCP()
@ -230,8 +218,7 @@ class TestToolDecorator:
mcp.add_tool(MyClass.add)
result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
assert isinstance(result[0], TextContent)
assert result[0].text == "3"
assert result[0].text == "3" # type: ignore[attr-defined]
async def test_tool_decorator_with_tags(self):
"""Test that the tool decorator properly sets tags."""
@ -262,8 +249,7 @@ class TestToolDecorator:
# Call the tool by its custom name
result = await mcp._mcp_call_tool("custom_multiply", {"a": 5, "b": 3})
assert isinstance(result[0], TextContent)
assert result[0].text == "15"
assert result[0].text == "15" # type: ignore[attr-defined]
# Original name should not be registered
assert "multiply" not in tools
@ -316,8 +302,7 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Hello, world!"
assert result[0].text == "Hello, world!" # type: ignore[attr-defined]
async def test_resource_decorator_incorrect_usage(self):
mcp = FastMCP()
@ -344,8 +329,7 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Hello, world!"
assert result[0].text == "Hello, world!" # type: ignore[attr-defined]
async def test_resource_decorator_with_description(self):
mcp = FastMCP()
@ -389,8 +373,7 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "My prefix: Hello, world!"
assert result[0].text == "My prefix: Hello, world!" # type: ignore[attr-defined]
async def test_resource_decorator_classmethod(self):
mcp = FastMCP()
@ -408,8 +391,7 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Class prefix: Hello, world!"
assert result[0].text == "Class prefix: Hello, world!" # type: ignore[attr-defined]
async def test_resource_decorator_staticmethod(self):
mcp = FastMCP()
@ -422,8 +404,7 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Static Hello, world!"
assert result[0].text == "Static Hello, world!" # type: ignore[attr-defined]
async def test_resource_decorator_async_function(self):
mcp = FastMCP()
@ -434,8 +415,7 @@ class TestResourceDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Async Hello, world!"
assert result[0].text == "Async Hello, world!" # type: ignore[attr-defined]
class TestTemplateDecorator:
@ -454,8 +434,7 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Data for test"
assert result[0].text == "Data for test" # type: ignore[attr-defined]
async def test_template_decorator_incorrect_usage(self):
mcp = FastMCP()
@ -482,8 +461,7 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Data for test"
assert result[0].text == "Data for test" # type: ignore[attr-defined]
async def test_template_decorator_with_description(self):
mcp = FastMCP()
@ -514,8 +492,7 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "My prefix: Data for test"
assert result[0].text == "My prefix: Data for test" # type: ignore[attr-defined]
async def test_template_decorator_classmethod(self):
mcp = FastMCP()
@ -535,8 +512,7 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Class prefix: Data for test"
assert result[0].text == "Class prefix: Data for test" # type: ignore[attr-defined]
async def test_template_decorator_staticmethod(self):
mcp = FastMCP()
@ -549,8 +525,7 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Static Data for test"
assert result[0].text == "Static Data for test" # type: ignore[attr-defined]
async def test_template_decorator_async_function(self):
mcp = FastMCP()
@ -561,8 +536,7 @@ class TestTemplateDecorator:
async with Client(mcp) as client:
result = await client.read_resource("resource://test/data")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Async Data for test"
assert result[0].text == "Async Data for test" # type: ignore[attr-defined]
async def test_template_decorator_with_tags(self):
"""Test that the template decorator properly sets tags."""
@ -603,8 +577,7 @@ class TestPromptDecorator:
assert prompt.name == "fn"
# Don't compare functions directly since validate_call wraps them
content = await prompt.render()
assert isinstance(content[0].content, TextContent)
assert content[0].content.text == "Hello, world!"
assert content[0].content.text == "Hello, world!" # type: ignore[attr-defined]
async def test_prompt_decorator_incorrect_usage(self):
mcp = FastMCP()
@ -629,8 +602,7 @@ class TestPromptDecorator:
prompt = prompts_dict["custom_name"]
assert prompt.name == "custom_name"
content = await prompt.render()
assert isinstance(content[0].content, TextContent)
assert content[0].content.text == "Hello, world!"
assert content[0].content.text == "Hello, world!" # type: ignore[attr-defined]
async def test_prompt_decorator_with_description(self):
mcp = FastMCP()
@ -644,8 +616,7 @@ class TestPromptDecorator:
prompt = prompts_dict["fn"]
assert prompt.description == "A custom description"
content = await prompt.render()
assert isinstance(content[0].content, TextContent)
assert content[0].content.text == "Hello, world!"
assert content[0].content.text == "Hello, world!" # type: ignore[attr-defined]
async def test_prompt_decorator_with_parameters(self):
mcp = FastMCP()
@ -668,16 +639,14 @@ class TestPromptDecorator:
result = await client.get_prompt("test_prompt", {"name": "World"})
assert len(result.messages) == 1
message = result.messages[0]
assert isinstance(message.content, TextContent)
assert message.content.text == "Hello, World!"
assert message.content.text == "Hello, World!" # type: ignore[attr-defined]
result = await client.get_prompt(
"test_prompt", {"name": "World", "greeting": "Hi"}
)
assert len(result.messages) == 1
message = result.messages[0]
assert isinstance(message.content, TextContent)
assert message.content.text == "Hi, World!"
assert message.content.text == "Hi, World!" # type: ignore[attr-defined]
async def test_prompt_decorator_instance_method(self):
mcp = FastMCP()
@ -696,8 +665,7 @@ class TestPromptDecorator:
result = await client.get_prompt("test_prompt")
assert len(result.messages) == 1
message = result.messages[0]
assert isinstance(message.content, TextContent)
assert message.content.text == "My prefix: Hello, world!"
assert message.content.text == "My prefix: Hello, world!" # type: ignore[attr-defined]
async def test_prompt_decorator_classmethod(self):
mcp = FastMCP()
@ -715,8 +683,7 @@ class TestPromptDecorator:
result = await client.get_prompt("test_prompt")
assert len(result.messages) == 1
message = result.messages[0]
assert isinstance(message.content, TextContent)
assert message.content.text == "Class prefix: Hello, world!"
assert message.content.text == "Class prefix: Hello, world!" # type: ignore[attr-defined]
async def test_prompt_decorator_staticmethod(self):
mcp = FastMCP()
@ -731,8 +698,7 @@ class TestPromptDecorator:
result = await client.get_prompt("test_prompt")
assert len(result.messages) == 1
message = result.messages[0]
assert isinstance(message.content, TextContent)
assert message.content.text == "Static Hello, world!"
assert message.content.text == "Static Hello, world!" # type: ignore[attr-defined]
async def test_prompt_decorator_async_function(self):
mcp = FastMCP()
@ -745,8 +711,7 @@ class TestPromptDecorator:
result = await client.get_prompt("test_prompt")
assert len(result.messages) == 1
message = result.messages[0]
assert isinstance(message.content, TextContent)
assert message.content.text == "Async Hello, world!"
assert message.content.text == "Async Hello, world!" # type: ignore[attr-defined]
async def test_prompt_decorator_with_tags(self):
"""Test that the prompt decorator properly sets tags."""
@ -943,20 +908,17 @@ class TestResourcePrefixMounting:
async with Client(main_server) as client:
# Regular resource
result = await client.read_resource("resource://prefix/test-resource")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Resource content"
assert result[0].text == "Resource content" # type: ignore[attr-defined]
# Absolute path resource
result = await client.read_resource("resource://prefix//absolute/path")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Absolute resource content"
assert result[0].text == "Absolute resource content" # type: ignore[attr-defined]
# Template resource
result = await client.read_resource(
"resource://prefix/param-value/template"
)
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource with param-value"
assert result[0].text == "Template resource with param-value" # type: ignore[attr-defined]
@pytest.mark.parametrize(
"uri,prefix,expected_match,expected_strip",
@ -1032,15 +994,12 @@ class TestResourcePrefixMounting:
# Verify we can access the resources
async with Client(target_server) as client:
result = await client.read_resource("resource://imported/test-resource")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Resource content"
assert result[0].text == "Resource content" # type: ignore[attr-defined]
result = await client.read_resource("resource://imported//absolute/path")
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Absolute resource content"
assert result[0].text == "Absolute resource content" # type: ignore[attr-defined]
result = await client.read_resource(
"resource://imported/param-value/template"
)
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource with param-value"
assert result[0].text == "Template resource with param-value" # type: ignore[attr-defined]

170
tests/server/test_server_interactions.py
View file

@ -10,7 +10,6 @@ import pydantic_core
import pytest
from mcp import McpError
from mcp.types import (
BlobResourceContents,
ImageContent,
TextContent,
TextResourceContents,
@ -77,14 +76,12 @@ class TestTools:
async def test_call_tool(self, tool_server: FastMCP):
async with Client(tool_server) as client:
result = await client.call_tool("add", {"x": 1, "y": 2})
assert isinstance(result[0], TextContent)
assert result[0].text == "3"
assert result[0].text == "3" # type: ignore[attr-defined]
async def test_call_tool_as_client(self, tool_server: FastMCP):
async with Client(tool_server) as client:
result = await client.call_tool("add", {"x": 1, "y": 2})
assert isinstance(result[0], TextContent)
assert result[0].text == "3"
assert result[0].text == "3" # type: ignore[attr-defined]
async def test_call_tool_error(self, tool_server: FastMCP):
async with Client(tool_server) as client:
@ -113,8 +110,7 @@ class TestTools:
async def test_tool_returns_list(self, tool_server: FastMCP):
async with Client(tool_server) as client:
result = await client.call_tool("list_tool", {})
assert isinstance(result[0], TextContent)
assert result[0].text == '[\n "x",\n 2\n]'
assert result[0].text == '[\n "x",\n 2\n]' # type: ignore[attr-defined]
class TestToolReturnTypes:
@ -127,8 +123,7 @@ class TestToolReturnTypes:
async with Client(mcp) as client:
result = await client.call_tool("string_tool", {})
assert isinstance(result[0], TextContent)
assert result[0].text == "Hello, world!"
assert result[0].text == "Hello, world!" # type: ignore[attr-defined]
async def test_bytes(self, tmp_path: Path):
mcp = FastMCP()
@ -139,8 +134,7 @@ class TestToolReturnTypes:
async with Client(mcp) as client:
result = await client.call_tool("bytes_tool", {})
assert isinstance(result[0], TextContent)
assert result[0].text == '"Hello, world!"'
assert result[0].text == '"Hello, world!"' # type: ignore[attr-defined]
async def test_uuid(self):
mcp = FastMCP()
@ -153,8 +147,7 @@ class TestToolReturnTypes:
async with Client(mcp) as client:
result = await client.call_tool("uuid_tool", {})
assert isinstance(result[0], TextContent)
assert result[0].text == pydantic_core.to_json(test_uuid).decode()
assert result[0].text == pydantic_core.to_json(test_uuid).decode() # type: ignore[attr-defined]
async def test_path(self):
mcp = FastMCP()
@ -167,8 +160,7 @@ class TestToolReturnTypes:
async with Client(mcp) as client:
result = await client.call_tool("path_tool", {})
assert isinstance(result[0], TextContent)
assert result[0].text == pydantic_core.to_json(test_path).decode()
assert result[0].text == pydantic_core.to_json(test_path).decode() # type: ignore[attr-defined]
async def test_datetime(self):
mcp = FastMCP()
@ -181,8 +173,7 @@ class TestToolReturnTypes:
async with Client(mcp) as client:
result = await client.call_tool("datetime_tool", {})
assert isinstance(result[0], TextContent)
assert result[0].text == pydantic_core.to_json(dt).decode()
assert result[0].text == pydantic_core.to_json(dt).decode() # type: ignore[attr-defined]
async def test_image(self, tmp_path: Path):
mcp = FastMCP()
@ -337,8 +328,7 @@ class TestToolParameters:
async with Client(mcp) as client:
# String with integer value should be coerced to int
result = await client.call_tool("add_one", {"x": "42"})
assert isinstance(result[0], TextContent)
assert result[0].text == "43"
assert result[0].text == "43" # type: ignore[attr-defined]
async def test_tool_bool_coercion(self):
"""Test string-to-bool type coercion."""
@ -351,12 +341,10 @@ class TestToolParameters:
async with Client(mcp) as client:
# String with boolean value should be coerced to bool
result = await client.call_tool("toggle", {"flag": "true"})
assert isinstance(result[0], TextContent)
assert result[0].text == "false"
assert result[0].text == "false" # type: ignore[attr-defined]
result = await client.call_tool("toggle", {"flag": "false"})
assert isinstance(result[0], TextContent)
assert result[0].text == "true"
assert result[0].text == "true" # type: ignore[attr-defined]
async def test_annotated_field_validation(self):
mcp = FastMCP()
@ -411,8 +399,7 @@ class TestToolParameters:
async with Client(mcp) as client:
result = await client.call_tool("analyze", {"x": "a"})
assert isinstance(result[0], TextContent)
assert result[0].text == "a"
assert result[0].text == "a" # type: ignore[attr-defined]
async def test_enum_type_validation_error(self):
mcp = FastMCP()
@ -444,8 +431,7 @@ class TestToolParameters:
async with Client(mcp) as client:
result = await client.call_tool("analyze", {"x": "red"})
assert isinstance(result[0], TextContent)
assert result[0].text == "red"
assert result[0].text == "red" # type: ignore[attr-defined]
async def test_union_type_validation(self):
mcp = FastMCP()
@ -456,12 +442,10 @@ class TestToolParameters:
async with Client(mcp) as client:
result = await client.call_tool("analyze", {"x": 1})
assert isinstance(result[0], TextContent)
assert result[0].text == "1"
assert result[0].text == "1" # type: ignore[attr-defined]
result = await client.call_tool("analyze", {"x": 1.0})
assert isinstance(result[0], TextContent)
assert result[0].text == "1.0"
assert result[0].text == "1.0" # type: ignore[attr-defined]
with pytest.raises(ToolError, match="Error calling tool 'analyze'"):
await client.call_tool("analyze", {"x": "not a number"})
@ -479,8 +463,7 @@ class TestToolParameters:
async with Client(mcp) as client:
result = await client.call_tool("send_path", {"path": str(test_path)})
assert isinstance(result[0], TextContent)
assert result[0].text == str(test_path)
assert result[0].text == str(test_path) # type: ignore[attr-defined]
async def test_path_type_error(self):
mcp = FastMCP()
@ -505,8 +488,7 @@ class TestToolParameters:
async with Client(mcp) as client:
result = await client.call_tool("send_uuid", {"x": test_uuid})
assert isinstance(result[0], TextContent)
assert result[0].text == str(test_uuid)
assert result[0].text == str(test_uuid) # type: ignore[attr-defined]
async def test_uuid_type_error(self):
mcp = FastMCP()
@ -530,8 +512,7 @@ class TestToolParameters:
async with Client(mcp) as client:
result = await client.call_tool("send_datetime", {"x": dt})
assert isinstance(result[0], TextContent)
assert result[0].text == dt.isoformat()
assert result[0].text == dt.isoformat() # type: ignore[attr-defined]
async def test_datetime_type_parse_string(self):
mcp = FastMCP()
@ -544,8 +525,7 @@ class TestToolParameters:
result = await client.call_tool(
"send_datetime", {"x": "2021-01-01T00:00:00"}
)
assert isinstance(result[0], TextContent)
assert result[0].text == "2021-01-01T00:00:00"
assert result[0].text == "2021-01-01T00:00:00" # type: ignore[attr-defined]
async def test_datetime_type_error(self):
mcp = FastMCP()
@ -567,8 +547,7 @@ class TestToolParameters:
async with Client(mcp) as client:
result = await client.call_tool("send_date", {"x": datetime.date.today()})
assert isinstance(result[0], TextContent)
assert result[0].text == datetime.date.today().isoformat()
assert result[0].text == datetime.date.today().isoformat() # type: ignore[attr-defined]
async def test_date_type_parse_string(self):
mcp = FastMCP()
@ -579,8 +558,7 @@ class TestToolParameters:
async with Client(mcp) as client:
result = await client.call_tool("send_date", {"x": "2021-01-01"})
assert isinstance(result[0], TextContent)
assert result[0].text == "2021-01-01"
assert result[0].text == "2021-01-01" # type: ignore[attr-defined]
async def test_timedelta_type(self):
mcp = FastMCP()
@ -593,8 +571,7 @@ class TestToolParameters:
result = await client.call_tool(
"send_timedelta", {"x": datetime.timedelta(days=1)}
)
assert isinstance(result[0], TextContent)
assert result[0].text == "1 day, 0:00:00"
assert result[0].text == "1 day, 0:00:00" # type: ignore[attr-defined]
async def test_timedelta_type_parse_int(self):
mcp = FastMCP()
@ -605,8 +582,7 @@ class TestToolParameters:
async with Client(mcp) as client:
result = await client.call_tool("send_timedelta", {"x": 1000})
assert isinstance(result[0], TextContent)
assert result[0].text == "0:16:40"
assert result[0].text == "0:16:40" # type: ignore[attr-defined]
class TestToolContextInjection:
@ -639,7 +615,7 @@ class TestToolContextInjection:
result = await client.call_tool("tool_with_context", {"x": 42})
assert len(result) == 1
content = result[0]
assert isinstance(content, TextContent)
assert content.text == "2" # type: ignore[attr-defined]
async def test_async_context(self):
"""Test that context works in async functions."""
@ -654,8 +630,7 @@ class TestToolContextInjection:
result = await client.call_tool("async_tool", {"x": 42})
assert len(result) == 1
content = result[0]
assert isinstance(content, TextContent)
assert content.text == "Async request 2: 42"
assert content.text == "Async request 2: 42" # type: ignore[attr-defined]
async def test_optional_context(self):
"""Test that context is optional."""
@ -669,8 +644,7 @@ class TestToolContextInjection:
result = await client.call_tool("no_context", {"x": 21})
assert len(result) == 1
content = result[0]
assert isinstance(content, TextContent)
assert content.text == "42"
assert content.text == "42" # type: ignore[attr-defined]
async def test_context_resource_access(self):
"""Test that context can access resources."""
@ -692,8 +666,7 @@ class TestToolContextInjection:
result = await client.call_tool("tool_with_resource", {})
assert len(result) == 1
content = result[0]
assert isinstance(content, TextContent)
assert "Read resource: resource data" in content.text
assert "Read resource: resource data" in content.text # type: ignore[attr-defined]
async def test_tool_decorator_with_tags(self):
"""Test that the tool decorator properly sets tags."""
@ -721,8 +694,7 @@ class TestToolContextInjection:
async with Client(mcp) as client:
result = await client.call_tool("MyTool", {"x": 2})
assert isinstance(result[0], TextContent)
assert result[0].text == "4"
assert result[0].text == "4" # type: ignore[attr-defined]
class TestResource:
@ -739,8 +711,7 @@ class TestResource:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Hello, world!"
assert result[0].text == "Hello, world!" # type: ignore[attr-defined]
async def test_binary_resource(self):
mcp = FastMCP()
@ -758,8 +729,7 @@ class TestResource:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://binary"))
assert isinstance(result[0], BlobResourceContents)
assert result[0].blob == base64.b64encode(b"Binary data").decode()
assert result[0].blob == base64.b64encode(b"Binary data").decode() # type: ignore[attr-defined]
async def test_file_resource_text(self, tmp_path: Path):
mcp = FastMCP()
@ -775,8 +745,7 @@ class TestResource:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("file://test.txt"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Hello from file!"
assert result[0].text == "Hello from file!" # type: ignore[attr-defined]
async def test_file_resource_binary(self, tmp_path: Path):
mcp = FastMCP()
@ -795,8 +764,7 @@ class TestResource:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("file://test.bin"))
assert isinstance(result[0], BlobResourceContents)
assert result[0].blob == base64.b64encode(b"Binary file data").decode()
assert result[0].blob == base64.b64encode(b"Binary file data").decode() # type: ignore[attr-defined]
class TestResourceContext:
@ -810,8 +778,7 @@ class TestResourceContext:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "2"
assert result[0].text == "2" # type: ignore[attr-defined]
class TestResourceTemplates:
@ -860,8 +827,7 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test/data"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Data for test"
assert result[0].text == "Data for test" # type: ignore[attr-defined]
async def test_resource_mismatched_params(self):
"""Test that mismatched parameters raise an error"""
@ -888,8 +854,7 @@ class TestResourceTemplates:
result = await client.read_resource(
AnyUrl("resource://cursor/fastmcp/data")
)
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Data for cursor/fastmcp"
assert result[0].text == "Data for cursor/fastmcp" # type: ignore[attr-defined]
async def test_resource_multiple_mismatched_params(self):
"""Test that mismatched parameters raise an error"""
@ -913,8 +878,7 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://static"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Static data"
assert result[0].text == "Static data" # type: ignore[attr-defined]
async def test_template_with_varkwargs(self):
"""Test that a template can have **kwargs."""
@ -926,8 +890,7 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("test://1/2/3"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "6"
assert result[0].text == "6" # type: ignore[attr-defined]
async def test_template_with_default_params(self):
"""Test that a template can have default parameters."""
@ -946,13 +909,11 @@ 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 isinstance(result[0], TextResourceContents)
assert result[0].text == "15" # 5 + default 10
assert result[0].text == "15" # type: ignore[attr-defined]
# Can also call with explicit params
result2 = await client.read_resource(AnyUrl("math://add/7"))
assert isinstance(result2[0], TextResourceContents)
assert result2[0].text == "17" # 7 + default 10
assert result2[0].text == "17" # type: ignore[attr-defined]
async def test_template_to_resource_conversion(self):
"""Test that a template can be converted to a resource."""
@ -971,8 +932,7 @@ 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 isinstance(result[0], TextResourceContents)
assert result[0].text == "Data for test"
assert result[0].text == "Data for test" # type: ignore[attr-defined]
async def test_stacked_resource_template_decorators(self):
"""Test that resource template decorators can be stacked."""
@ -1011,15 +971,15 @@ class TestResourceTemplates:
email_result = await client.read_resource(
AnyUrl("users://email/user@example.com")
)
assert isinstance(email_result[0], TextResourceContents)
email_data = json.loads(email_result[0].text)
assert email_result[0].text # type: ignore[attr-defined]
email_data = json.loads(email_result[0].text) # type: ignore[attr-defined]
assert email_data["lookup"] == "email"
assert email_data["email"] == "user@example.com"
# Test lookup by name
name_result = await client.read_resource(AnyUrl("users://name/John"))
assert isinstance(name_result[0], TextResourceContents)
name_data = json.loads(name_result[0].text)
assert name_result[0].text # type: ignore[attr-defined]
name_data = json.loads(name_result[0].text) # type: ignore[attr-defined]
assert name_data["lookup"] == "name"
assert name_data["name"] == "John"
assert name_data["email"] == "dummy@example.com"
@ -1044,8 +1004,7 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test/data"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource: test/data"
assert result[0].text == "Template resource: test/data" # type: ignore[attr-defined]
async def test_templates_match_in_order_of_definition(self):
"""
@ -1065,12 +1024,10 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://a/b/c"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource 1: a/b/c"
assert result[0].text == "Template resource 1: a/b/c" # type: ignore[attr-defined]
result = await client.read_resource(AnyUrl("resource://a/b"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource 1: a/b"
assert result[0].text == "Template resource 1: a/b" # type: ignore[attr-defined]
async def test_templates_shadow_each_other_reorder(self):
"""
@ -1089,12 +1046,10 @@ class TestResourceTemplates:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://a/b/c"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource 2: a/b/c"
assert result[0].text == "Template resource 2: a/b/c" # type: ignore[attr-defined]
result = await client.read_resource(AnyUrl("resource://a/b"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text == "Template resource 1: a/b"
assert result[0].text == "Template resource 1: a/b" # type: ignore[attr-defined]
class TestResourceTemplateContext:
@ -1108,8 +1063,7 @@ class TestResourceTemplateContext:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text.startswith("Resource template: test 2")
assert result[0].text.startswith("Resource template: test 2") # type: ignore[attr-defined]
async def test_resource_template_context_with_callable_object(self):
mcp = FastMCP()
@ -1122,8 +1076,7 @@ class TestResourceTemplateContext:
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("resource://test"))
assert isinstance(result[0], TextResourceContents)
assert result[0].text.startswith("Resource template: test 2")
assert result[0].text.startswith("Resource template: test 2") # type: ignore[attr-defined]
class TestPrompts:
@ -1143,8 +1096,7 @@ class TestPrompts:
assert prompt.name == "fn"
# Don't compare functions directly since validate_call wraps them
content = await prompt.render()
assert isinstance(content[0].content, TextContent)
assert content[0].content.text == "Hello, world!"
assert content[0].content.text == "Hello, world!" # type: ignore[attr-defined]
async def test_prompt_decorator_with_name(self):
"""Test prompt decorator with custom name."""
@ -1159,8 +1111,7 @@ class TestPrompts:
prompt = prompts_dict["custom_name"]
assert prompt.name == "custom_name"
content = await prompt.render()
assert isinstance(content[0].content, TextContent)
assert content[0].content.text == "Hello, world!"
assert content[0].content.text == "Hello, world!" # type: ignore[attr-defined]
async def test_prompt_decorator_with_description(self):
"""Test prompt decorator with custom description."""
@ -1175,8 +1126,7 @@ class TestPrompts:
prompt = prompts_dict["fn"]
assert prompt.description == "A custom description"
content = await prompt.render()
assert isinstance(content[0].content, TextContent)
assert content[0].content.text == "Hello, world!"
assert content[0].content.text == "Hello, world!" # type: ignore[attr-defined]
def test_prompt_decorator_error(self):
"""Test error when decorator is used incorrectly."""
@ -1224,8 +1174,7 @@ class TestPrompts:
message = result.messages[0]
assert message.role == "user"
content = message.content
assert isinstance(content, TextContent)
assert content.text == "Hello, World!"
assert content.text == "Hello, World!" # type: ignore[attr-defined]
async def test_get_prompt_with_resource(self):
"""Test getting a prompt that returns resource content."""
@ -1249,10 +1198,10 @@ class TestPrompts:
result = await client.get_prompt("fn")
assert result.messages[0].role == "user"
content = result.messages[0].content
assert isinstance(content, EmbeddedResource)
assert isinstance(content, EmbeddedResource) # type: ignore[attr-defined]
resource = content.resource
assert isinstance(resource, TextResourceContents)
assert resource.text == "File contents"
assert isinstance(resource, TextResourceContents) # type: ignore[attr-defined]
assert resource.text == "File contents" # type: ignore[attr-defined]
assert resource.mimeType == "text/plain"
async def test_get_unknown_prompt(self):
@ -1342,5 +1291,4 @@ class TestPromptContext:
assert len(result.messages) == 1
message = result.messages[0]
assert message.role == "user"
assert isinstance(message.content, TextContent)
assert message.content.text == "Hello, World! 2"
assert message.content.text == "Hello, World! 2" # type: ignore[attr-defined]

7
tests/server/test_tool_annotations.py
View file

@ -1,6 +1,6 @@
from typing import Any
from mcp.types import TextContent, ToolAnnotations
from mcp.types import ToolAnnotations
from fastmcp import Client, FastMCP
@ -212,8 +212,7 @@ async def test_tool_functionality_with_annotations():
"create_item", {"name": "test_item", "value": 42}
)
assert len(result) == 1
assert isinstance(result[0], TextContent)
# The result should contain the expected JSON
assert '"name": "test_item"' in result[0].text
assert '"value": 42' in result[0].text
assert '"name": "test_item"' in result[0].text # type: ignore[attr-defined]
assert '"value": 42' in result[0].text # type: ignore[attr-defined]

34
tests/test_examples.py
View file

@ -1,10 +1,5 @@
"""Tests for example servers"""
from mcp.types import (
PromptMessage,
TextContent,
TextResourceContents,
)
from pydantic import AnyUrl
from fastmcp import Client
@ -17,8 +12,7 @@ async def test_simple_echo():
async with Client(mcp) as client:
result = await client.call_tool("echo", {"text": "hello"})
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "hello"
assert result[0].text == "hello" # type: ignore[attr-defined]
async def test_complex_inputs():
@ -31,8 +25,7 @@ async def test_complex_inputs():
"name_shrimp", {"tank": tank, "extra_names": ["charlie"]}
)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == '[\n "bob",\n "alice",\n "charlie"\n]'
assert result[0].text == '[\n "bob",\n "alice",\n "charlie"\n]' # type: ignore[attr-defined]
async def test_desktop(monkeypatch):
@ -43,15 +36,12 @@ async def test_desktop(monkeypatch):
# Test the add function
result = await client.call_tool("add", {"a": 1, "b": 2})
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "3"
assert result[0].text == "3" # type: ignore[attr-defined]
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("greeting://rooter12"))
assert len(result) == 1
assert isinstance(result[0], TextResourceContents)
assert isinstance(result[0].text, str)
assert result[0].text == "Hello, rooter12!"
assert result[0].text == "Hello, rooter12!" # type: ignore[attr-defined]
async def test_echo():
@ -61,27 +51,19 @@ async def test_echo():
async with Client(mcp) as client:
result = await client.call_tool("echo_tool", {"text": "hello"})
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "hello"
assert result[0].text == "hello" # type: ignore[attr-defined]
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("echo://static"))
assert len(result) == 1
assert isinstance(result[0], TextResourceContents)
assert isinstance(result[0].text, str)
assert result[0].text == "Echo!"
assert result[0].text == "Echo!" # type: ignore[attr-defined]
async with Client(mcp) as client:
result = await client.read_resource(AnyUrl("echo://server42"))
assert len(result) == 1
assert isinstance(result[0], TextResourceContents)
assert isinstance(result[0].text, str)
assert result[0].text == "Echo: server42"
assert result[0].text == "Echo: server42" # type: ignore[attr-defined]
async with Client(mcp) as client:
result = await client.get_prompt("echo", {"text": "hello"})
assert len(result.messages) == 1
assert isinstance(result.messages[0], PromptMessage)
assert isinstance(result.messages[0].content, TextContent)
assert isinstance(result.messages[0].content.text, str)
assert result.messages[0].content.text == "hello"
assert result.messages[0].content.text == "hello" # type: ignore[attr-defined]

46
tests/tools/test_tool.py
View file

@ -1,5 +1,5 @@
import pytest
from mcp.types import ImageContent, TextContent
from mcp.types import ImageContent
from pydantic import BaseModel
from fastmcp import FastMCP, Image
@ -209,9 +209,7 @@ class TestLegacyToolJsonParsing:
# Run the tool which will do JSON parsing
result = await tool.run(json_args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "1-a,b,c"
assert result[0].text == "1-a,b,c" # type: ignore[attr-dict]
async def test_str_vs_list_str(self):
"""Test handling of string vs list[str] type annotations."""
@ -223,23 +221,17 @@ class TestLegacyToolJsonParsing:
# Test regular string input (should remain a string)
result = await tool.run({"str_or_list": "hello"})
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "hello"
assert result[0].text == "hello" # type: ignore[attr-dict]
# Test JSON string input (should be parsed as a string)
result = await tool.run({"str_or_list": '"hello"'})
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "hello"
assert result[0].text == "hello" # type: ignore[attr-dict]
# Test JSON list input (should be parsed as a list)
result = await tool.run({"str_or_list": '["hello", "world"]'})
assert len(result) == 1
assert isinstance(result[0], TextContent)
# The exact formatting might vary, so we just check that it contains the key elements
text_without_whitespace = result[0].text.replace(" ", "").replace("\n", "")
text_without_whitespace = result[0].text.replace(" ", "").replace("\n", "") # type: ignore[attr-dict]
assert "hello" in text_without_whitespace
assert "world" in text_without_whitespace
assert "[" in text_without_whitespace
@ -256,9 +248,7 @@ class TestLegacyToolJsonParsing:
# Invalid JSON should remain a string
invalid_json = "{'nice to meet you': 'hello', 'goodbye': 5}"
result = await tool.run({"string": invalid_json})
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == invalid_json
assert result[0].text == invalid_json # type: ignore[attr-dict]
async def test_keep_str_union_as_str(self):
"""Test that string arguments are kept as strings when parsing would create an invalid value"""
@ -273,9 +263,7 @@ class TestLegacyToolJsonParsing:
# Invalid JSON for the union type should remain a string
invalid_json = "{'nice to meet you': 'hello', 'goodbye': 5}"
result = await tool.run({"string": invalid_json})
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == invalid_json
assert result[0].text == invalid_json # type: ignore[attr-dict]
async def test_complex_type_validation(self):
"""Test that parsed JSON is validated against complex types"""
@ -292,11 +280,9 @@ class TestLegacyToolJsonParsing:
# Valid JSON for the model
valid_json = '{"x": 1, "y": {"1": "hello"}}'
result = await tool.run({"data": valid_json})
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert '"x": 1' in result[0].text
assert '"y": {' in result[0].text
assert '"1": "hello"' in result[0].text
assert '"x": 1' in result[0].text # type: ignore[attr-dict]
assert '"y": {' in result[0].text # type: ignore[attr-dict]
assert '"1": "hello"' in result[0].text # type: ignore[attr-dict]
# Invalid JSON for the model (y has string keys, not int keys)
# Should throw a validation error
@ -317,8 +303,7 @@ class TestLegacyToolJsonParsing:
result = await client.call_tool(
"process_list", {"items": "[1, 2, 3, 4, 5]"}
)
assert isinstance(result[0], TextContent)
assert result[0].text == "15"
assert result[0].text == "15" # type: ignore[attr-dict]
async def test_tool_list_coercion_error(self):
"""Test that a list coercion error is raised if the input is not a valid list."""
@ -348,8 +333,7 @@ class TestLegacyToolJsonParsing:
result = await client.call_tool(
"process_dict", {"data": '{"a": 1, "b": "2", "c": 3}'}
)
assert isinstance(result[0], TextContent)
assert result[0].text == "6"
assert result[0].text == "6" # type: ignore[attr-dict]
async def test_tool_set_coercion(self):
"""Test JSON string to set type coercion."""
@ -362,8 +346,7 @@ class TestLegacyToolJsonParsing:
async with Client(mcp) as client:
result = await client.call_tool("process_set", {"items": "[1, 2, 3, 4, 5]"})
assert isinstance(result[0], TextContent)
assert result[0].text == "15"
assert result[0].text == "15" # type: ignore[attr-dict]
async def test_tool_tuple_coercion(self):
"""Test JSON string to tuple type coercion."""
@ -376,5 +359,4 @@ class TestLegacyToolJsonParsing:
async with Client(mcp) as client:
result = await client.call_tool("process_tuple", {"items": '["1", "two"]'})
assert isinstance(result[0], TextContent)
assert result[0].text == "4"
assert result[0].text == "4" # type: ignore[attr-dict]

111
tests/tools/test_tool_manager.py
View file

@ -5,7 +5,7 @@ from typing import Annotated, Any
import pydantic_core
import pytest
from mcp.types import ImageContent, TextContent
from mcp.types import ImageContent
from pydantic import BaseModel
from fastmcp import Context, FastMCP, Image
@ -318,13 +318,8 @@ class TestCallTools:
manager = ToolManager()
manager.add_tool_from_fn(add)
result = await manager.call_tool("add", {"a": 1, "b": 2})
assert isinstance(result, list)
assert len(result) == 1
from mcp.types import TextContent
assert isinstance(result[0], TextContent)
assert result[0].text == "3"
assert json.loads(result[0].text) == 3
assert result[0].text == "3" # type: ignore[attr-defined]
async def test_call_async_tool(self):
async def double(n: int) -> int:
@ -334,12 +329,7 @@ class TestCallTools:
manager = ToolManager()
manager.add_tool_from_fn(double)
result = await manager.call_tool("double", {"n": 5})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "10"
assert json.loads(result[0].text) == 10
assert result[0].text == "10" # type: ignore[attr-defined]
async def test_call_tool_callable_object(self):
class Adder:
@ -352,11 +342,7 @@ class TestCallTools:
manager = ToolManager()
manager.add_tool_from_fn(Adder())
result = await manager.call_tool("Adder", {"x": 1, "y": 2})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "3"
assert json.loads(result[0].text) == 3
assert result[0].text == "3" # type: ignore[attr-defined]
async def test_call_tool_callable_object_async(self):
class Adder:
@ -369,11 +355,7 @@ class TestCallTools:
manager = ToolManager()
manager.add_tool_from_fn(Adder())
result = await manager.call_tool("Adder", {"x": 1, "y": 2})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "3"
assert json.loads(result[0].text) == 3
assert result[0].text == "3" # type: ignore[attr-defined]
async def test_call_tool_with_default_args(self):
def add(a: int, b: int = 1) -> int:
@ -383,12 +365,8 @@ class TestCallTools:
manager = ToolManager()
manager.add_tool_from_fn(add)
result = await manager.call_tool("add", {"a": 1})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "2"
assert json.loads(result[0].text) == 2
assert result[0].text == "2" # type: ignore[attr-defined]
async def test_call_tool_with_missing_args(self):
def add(a: int, b: int) -> int:
@ -413,11 +391,7 @@ class TestCallTools:
manager.add_tool_from_fn(sum_vals)
result = await manager.call_tool("sum_vals", {"vals": [1, 2, 3]})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "6"
assert json.loads(result[0].text) == 6
assert result[0].text == "6" # type: ignore[attr-defined]
async def test_call_tool_with_list_int_input_legacy_behavior(self):
"""Legacy behavior -- parse a stringified JSON object"""
@ -431,11 +405,7 @@ class TestCallTools:
with temporary_settings(tool_attempt_parse_json_args=True):
result = await manager.call_tool("sum_vals", {"vals": "[1, 2, 3]"})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "6"
assert json.loads(result[0].text) == 6
assert result[0].text == "6" # type: ignore[attr-defined]
async def test_call_tool_with_list_str_or_str_input(self):
def concat_strs(vals: list[str] | str) -> str:
@ -446,16 +416,10 @@ class TestCallTools:
# Try both with plain python object and with JSON list
result = await manager.call_tool("concat_strs", {"vals": ["a", "b", "c"]})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "abc"
assert result[0].text == "abc" # type: ignore[attr-defined]
result = await manager.call_tool("concat_strs", {"vals": "a"})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "a"
assert result[0].text == "a" # type: ignore[attr-defined]
async def test_call_tool_with_list_str_or_str_input_legacy_behavior(self):
"""Legacy behavior -- parse a stringified JSON object"""
@ -468,16 +432,10 @@ class TestCallTools:
with temporary_settings(tool_attempt_parse_json_args=True):
result = await manager.call_tool("concat_strs", {"vals": '["a", "b", "c"]'})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "abc"
assert result[0].text == "abc" # type: ignore[attr-defined]
result = await manager.call_tool("concat_strs", {"vals": '"a"'})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "a"
assert result[0].text == "a" # type: ignore[attr-defined]
async def test_call_tool_with_complex_model(self):
class MyShrimpTank(BaseModel):
@ -507,10 +465,7 @@ class TestCallTools:
},
)
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == '[\n "rex",\n "gertrude"\n]'
assert result[0].text == '[\n "rex",\n "gertrude"\n]' # type: ignore[attr-defined]
async def test_call_tool_with_custom_serializer(self):
"""Test that a custom serializer provided to FastMCP is used by tools."""
@ -530,10 +485,7 @@ class TestCallTools:
manager.add_tool_from_fn(get_data)
result = await manager.call_tool("get_data", {})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == 'CUSTOM:{"key": "value", "number": 123}'
assert result[0].text == 'CUSTOM:{"key": "value", "number": 123}' # type: ignore[attr-defined]
async def test_call_tool_with_list_result_custom_serializer(self):
"""Test that a custom serializer provided to FastMCP is used by tools that return lists."""
@ -555,12 +507,9 @@ class TestCallTools:
manager.add_tool_from_fn(get_data)
result = await manager.call_tool("get_data", {})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert (
result[0].text
== 'CUSTOM:[{"key": "value", "number": 123}, {"key": "value2", "number": 456}]'
result[0].text # type: ignore[attr-defined]
== 'CUSTOM:[{"key": "value", "number": 123}, {"key": "value2", "number": 456}]' # type: ignore[attr-defined]
)
async def test_custom_serializer_fallback_on_error(self):
@ -580,10 +529,7 @@ class TestCallTools:
manager.add_tool_from_fn(get_data)
result = await manager.call_tool("get_data", {})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == pydantic_core.to_json(uuid_result).decode()
assert result[0].text == pydantic_core.to_json(uuid_result).decode() # type: ignore[attr-defined]
class TestToolSchema:
@ -648,10 +594,7 @@ class TestContextHandling:
with context:
result = await manager.call_tool("tool_with_context", {"x": 42})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "42"
assert result[0].text == "42" # type: ignore[attr-defined]
async def test_context_injection_async(self):
"""Test that context is properly injected in async tools."""
@ -668,14 +611,10 @@ class TestContextHandling:
with context:
result = await manager.call_tool("async_tool", {"x": 42})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "42"
assert result[0].text == "42" # type: ignore[attr-defined]
async def test_context_optional(self):
"""Test that context is optional when calling tools."""
from mcp.types import TextContent
def tool_with_context(x: int, ctx: Context | None) -> int:
return x
@ -689,10 +628,7 @@ class TestContextHandling:
with context:
result = await manager.call_tool("tool_with_context", {"x": 42})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "42"
assert result[0].text == "42" # type: ignore[attr-defined]
def test_parameterized_context_parameter_detection(self):
"""Test that context parameters are properly detected in
@ -782,7 +718,6 @@ class TestCustomToolNames:
async def test_call_tool_with_custom_name(self):
"""Test calling a tool added with a custom name."""
from mcp.types import TextContent
def multiply(a: int, b: int) -> int:
"""Multiply two numbers."""
@ -793,11 +728,7 @@ class TestCustomToolNames:
# Tool should be callable by its custom name
result = await manager.call_tool("custom_multiply", {"a": 5, "b": 3})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "15"
assert json.loads(result[0].text) == 15
assert result[0].text == "15" # type: ignore[attr-defined]
# Original name should not be registered
with pytest.raises(NotFoundError, match="Unknown tool: multiply"):

8
tests/utilities/test_mcp_config.py
View file

@ -1,8 +1,6 @@
import inspect
from pathlib import Path
from mcp.types import TextContent
from fastmcp.client.client import Client
from fastmcp.client.transports import (
SSETransport,
@ -136,7 +134,5 @@ async def test_multi_client(tmp_path: Path):
result_1 = await client.call_tool("test_1_add", {"a": 1, "b": 2})
result_2 = await client.call_tool("test_2_add", {"a": 1, "b": 2})
assert isinstance(result_1[0], TextContent)
assert result_1[0].text == "3"
assert isinstance(result_2[0], TextContent)
assert result_2[0].text == "3"
assert result_1[0].text == "3" # type: ignore[attr-dict]
assert result_2[0].text == "3" # type: ignore[attr-dict]