modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: use snake case instead of camel case in types (#1894)

Browse source
This commit is contained in:
Marcelo Trylesinski 2026-01-16 15:51:27 +01:00 • committed by GitHub
commit e94b386a13
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 1063 additions and 1057 deletions
Download patch file Download diff file Expand all files Collapse all files

10
tests/server/fastmcp/prompts/test_base.py
View file

@ -96,7 +96,7 @@ class TestRenderPrompt:
resource=TextResourceContents(
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
mime_type="text/plain",
),
)
)
@ -109,7 +109,7 @@ class TestRenderPrompt:
resource=TextResourceContents(
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
mime_type="text/plain",
),
)
)
@ -128,7 +128,7 @@ class TestRenderPrompt:
resource=TextResourceContents(
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
mime_type="text/plain",
),
)
),
@ -144,7 +144,7 @@ class TestRenderPrompt:
resource=TextResourceContents(
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
mime_type="text/plain",
),
)
),
@ -176,7 +176,7 @@ class TestRenderPrompt:
resource=TextResourceContents(
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
mime_type="text/plain",
),
)
)

2
tests/server/fastmcp/test_elicitation.py
View file

@ -290,7 +290,7 @@ async def test_elicitation_with_default_values():
async def callback_schema_verify(context: RequestContext[ClientSession, None], params: ElicitRequestParams):
# Verify the schema includes defaults
assert isinstance(params, types.ElicitRequestFormParams), "Expected form mode elicitation"
schema = params.requestedSchema
schema = params.requested_schema
props = schema["properties"]
assert props["name"]["default"] == "Guest"

4
tests/server/fastmcp/test_func_metadata.py
View file

@ -850,7 +850,7 @@ def test_tool_call_result_annotated_is_structured_and_converted():
name: str
def func_returning_annotated_tool_call_result() -> Annotated[CallToolResult, PersonClass]: # pragma: no cover
return CallToolResult(content=[], structuredContent={"name": "Brandon"})
return CallToolResult(content=[], structured_content={"name": "Brandon"})
meta = func_metadata(func_returning_annotated_tool_call_result)
@ -870,7 +870,7 @@ def test_tool_call_result_annotated_is_structured_and_invalid():
name: str
def func_returning_annotated_tool_call_result() -> Annotated[CallToolResult, PersonClass]: # pragma: no cover
return CallToolResult(content=[], structuredContent={"person": "Brandon"})
return CallToolResult(content=[], structured_content={"person": "Brandon"})
meta = func_metadata(func_returning_annotated_tool_call_result)

20
tests/server/fastmcp/test_integration.py
View file

@ -258,7 +258,7 @@ async def test_basic_tools(server_transport: str, server_url: str) -> None:
# Test initialization
result = await session.initialize()
assert isinstance(result, InitializeResult)
assert result.serverInfo.name == "Tool Example"
assert result.server_info.name == "Tool Example"
assert result.capabilities.tools is not None
# Test sum tool
@ -295,7 +295,7 @@ async def test_basic_resources(server_transport: str, server_url: str) -> None:
# Test initialization
result = await session.initialize()
assert isinstance(result, InitializeResult)
assert result.serverInfo.name == "Resource Example"
assert result.server_info.name == "Resource Example"
assert result.capabilities.resources is not None
# Test document resource
@ -336,7 +336,7 @@ async def test_basic_prompts(server_transport: str, server_url: str) -> None:
# Test initialization
result = await session.initialize()
assert isinstance(result, InitializeResult)
assert result.serverInfo.name == "Prompt Example"
assert result.server_info.name == "Prompt Example"
assert result.capabilities.prompts is not None
# Test review_code prompt
@ -396,7 +396,7 @@ async def test_tool_progress(server_transport: str, server_url: str) -> None:
# Test initialization
result = await session.initialize()
assert isinstance(result, InitializeResult)
assert result.serverInfo.name == "Progress Example"
assert result.server_info.name == "Progress Example"
# Test progress callback
progress_updates = []
@ -449,7 +449,7 @@ async def test_sampling(server_transport: str, server_url: str) -> None:
# Test initialization
result = await session.initialize()
assert isinstance(result, InitializeResult)
assert result.serverInfo.name == "Sampling Example"
assert result.server_info.name == "Sampling Example"
assert result.capabilities.tools is not None
# Test sampling tool
@ -480,7 +480,7 @@ async def test_elicitation(server_transport: str, server_url: str) -> None:
# Test initialization
result = await session.initialize()
assert isinstance(result, InitializeResult)
assert result.serverInfo.name == "Elicitation Example"
assert result.server_info.name == "Elicitation Example"
# Test booking with unavailable date (triggers elicitation)
booking_result = await session.call_tool(
@ -537,7 +537,7 @@ async def test_notifications(server_transport: str, server_url: str) -> None:
# Test initialization
result = await session.initialize()
assert isinstance(result, InitializeResult)
assert result.serverInfo.name == "Notifications Example"
assert result.server_info.name == "Notifications Example"
# Call tool that generates notifications
tool_result = await session.call_tool("process_data", {"data": "test_data"})
@ -578,7 +578,7 @@ async def test_completion(server_transport: str, server_url: str) -> None:
# Test initialization
result = await session.initialize()
assert isinstance(result, InitializeResult)
assert result.serverInfo.name == "Example"
assert result.server_info.name == "Example"
assert result.capabilities.resources is not None
assert result.capabilities.prompts is not None
@ -635,7 +635,7 @@ async def test_fastmcp_quickstart(server_transport: str, server_url: str) -> Non
# Test initialization
result = await session.initialize()
assert isinstance(result, InitializeResult)
assert result.serverInfo.name == "Demo"
assert result.server_info.name == "Demo"
# Test add tool
tool_result = await session.call_tool("add", {"a": 10, "b": 20})
@ -673,7 +673,7 @@ async def test_structured_output(server_transport: str, server_url: str) -> None
# Test initialization
result = await session.initialize()
assert isinstance(result, InitializeResult)
assert result.serverInfo.name == "Structured Output Example"
assert result.server_info.name == "Structured Output Example"
# Test get_weather tool
weather_result = await session.call_tool("get_weather", {"city": "New York"})

2
tests/server/fastmcp/test_parameter_descriptions.py
View file

@ -23,7 +23,7 @@ async def test_parameter_descriptions():
tool = tools[0]
# Check that parameter descriptions are present in the schema
properties = tool.inputSchema["properties"]
properties = tool.input_schema["properties"]
assert "name" in properties
assert properties["name"]["description"] == "The name to greet"
assert "title" in properties

117
tests/server/fastmcp/test_server.py
View file

@ -36,7 +36,7 @@ class TestServer:
instructions="Server instructions",
website_url="https://example.com/mcp_server",
version="1.0",
icons=[Icon(src="https://example.com/icon.png", mimeType="image/png", sizes=["48x48", "96x96"])],
icons=[Icon(src="https://example.com/icon.png", mime_type="image/png", sizes=["48x48", "96x96"])],
)
assert mcp.name == "FastMCP"
assert mcp.title == "FastMCP Server"
@ -197,8 +197,8 @@ def audio_tool_fn(path: str) -> Audio:
def mixed_content_tool_fn() -> list[ContentBlock]:
return [
TextContent(type="text", text="Hello"),
ImageContent(type="image", data="abc", mimeType="image/png"),
AudioContent(type="audio", data="def", mimeType="audio/wav"),
ImageContent(type="image", data="abc", mime_type="image/png"),
AudioContent(type="audio", data="def", mime_type="audio/wav"),
]
@ -237,7 +237,7 @@ class TestServerTools:
content = result.content[0]
assert isinstance(content, TextContent)
assert "Test error" in content.text
assert result.isError is True
assert result.is_error is True
@pytest.mark.anyio
async def test_tool_error_handling(self):
@ -249,7 +249,7 @@ class TestServerTools:
content = result.content[0]
assert isinstance(content, TextContent)
assert "Test error" in content.text
assert result.isError is True
assert result.is_error is True
@pytest.mark.anyio
async def test_tool_error_details(self):
@ -262,7 +262,7 @@ class TestServerTools:
assert isinstance(content, TextContent)
assert isinstance(content.text, str)
assert "Test error" in content.text
assert result.isError is True
assert result.is_error is True
@pytest.mark.anyio
async def test_tool_return_value_conversion(self):
@ -275,8 +275,8 @@ class TestServerTools:
assert isinstance(content, TextContent)
assert content.text == "3"
# Check structured content - int return type should have structured output
assert result.structuredContent is not None
assert result.structuredContent == {"result": 3}
assert result.structured_content is not None
assert result.structured_content == {"result": 3}
@pytest.mark.anyio
async def test_tool_image_helper(self, tmp_path: Path):
@ -292,12 +292,12 @@ class TestServerTools:
content = result.content[0]
assert isinstance(content, ImageContent)
assert content.type == "image"
assert content.mimeType == "image/png"
assert content.mime_type == "image/png"
# Verify base64 encoding
decoded = base64.b64decode(content.data)
assert decoded == b"fake png data"
# Check structured content - Image return type should NOT have structured output
assert result.structuredContent is None
assert result.structured_content is None
@pytest.mark.anyio
async def test_tool_audio_helper(self, tmp_path: Path):
@ -313,12 +313,12 @@ class TestServerTools:
content = result.content[0]
assert isinstance(content, AudioContent)
assert content.type == "audio"
assert content.mimeType == "audio/wav"
assert content.mime_type == "audio/wav"
# Verify base64 encoding
decoded = base64.b64decode(content.data)
assert decoded == b"fake wav data"
# Check structured content - Image return type should NOT have structured output
assert result.structuredContent is None
assert result.structured_content is None
@pytest.mark.parametrize(
"filename,expected_mime_type",
@ -348,7 +348,7 @@ class TestServerTools:
content = result.content[0]
assert isinstance(content, AudioContent)
assert content.type == "audio"
assert content.mimeType == expected_mime_type
assert content.mime_type == expected_mime_type
# Verify base64 encoding
decoded = base64.b64decode(content.data)
assert decoded == b"fake audio data"
@ -364,14 +364,14 @@ class TestServerTools:
assert isinstance(content1, TextContent)
assert content1.text == "Hello"
assert isinstance(content2, ImageContent)
assert content2.mimeType == "image/png"
assert content2.mime_type == "image/png"
assert content2.data == "abc"
assert isinstance(content3, AudioContent)
assert content3.mimeType == "audio/wav"
assert content3.mime_type == "audio/wav"
assert content3.data == "def"
assert result.structuredContent is not None
assert "result" in result.structuredContent
structured_result = result.structuredContent["result"]
assert result.structured_content is not None
assert "result" in result.structured_content
structured_result = result.structured_content["result"]
assert len(structured_result) == 3
expected_content = [
@ -419,12 +419,12 @@ class TestServerTools:
# Check image conversion
content2 = result.content[1]
assert isinstance(content2, ImageContent)
assert content2.mimeType == "image/png"
assert content2.mime_type == "image/png"
assert base64.b64decode(content2.data) == b"test image data"
# Check audio conversion
content3 = result.content[2]
assert isinstance(content3, AudioContent)
assert content3.mimeType == "audio/wav"
assert content3.mime_type == "audio/wav"
assert base64.b64decode(content3.data) == b"test audio data"
# Check dict conversion
content4 = result.content[3]
@ -435,7 +435,7 @@ class TestServerTools:
assert isinstance(content5, TextContent)
assert content5.text == "direct content"
# Check structured content - untyped list with Image objects should NOT have structured output
assert result.structuredContent is None
assert result.structured_content is None
@pytest.mark.anyio
async def test_tool_structured_output_basemodel(self):
@ -457,16 +457,16 @@ class TestServerTools:
# Check that the tool has outputSchema
tools = await client.list_tools()
tool = next(t for t in tools.tools if t.name == "get_user")
assert tool.outputSchema is not None
assert tool.outputSchema["type"] == "object"
assert "name" in tool.outputSchema["properties"]
assert "age" in tool.outputSchema["properties"]
assert tool.output_schema is not None
assert tool.output_schema["type"] == "object"
assert "name" in tool.output_schema["properties"]
assert "age" in tool.output_schema["properties"]
# Call the tool and check structured output
result = await client.call_tool("get_user", {"user_id": 123})
assert result.isError is False
assert result.structuredContent is not None
assert result.structuredContent == {"name": "John Doe", "age": 30, "active": True}
assert result.is_error is False
assert result.structured_content is not None
assert result.structured_content == {"name": "John Doe", "age": 30, "active": True}
# Content should be JSON serialized version
assert len(result.content) == 1
assert isinstance(result.content[0], TextContent)
@ -487,17 +487,17 @@ class TestServerTools:
# Check that the tool has outputSchema
tools = await client.list_tools()
tool = next(t for t in tools.tools if t.name == "calculate_sum")
assert tool.outputSchema is not None
assert tool.output_schema is not None
# Primitive types are wrapped
assert tool.outputSchema["type"] == "object"
assert "result" in tool.outputSchema["properties"]
assert tool.outputSchema["properties"]["result"]["type"] == "integer"
assert tool.output_schema["type"] == "object"
assert "result" in tool.output_schema["properties"]
assert tool.output_schema["properties"]["result"]["type"] == "integer"
# Call the tool
result = await client.call_tool("calculate_sum", {"a": 5, "b": 7})
assert result.isError is False
assert result.structuredContent is not None
assert result.structuredContent == {"result": 12}
assert result.is_error is False
assert result.structured_content is not None
assert result.structured_content == {"result": 12}
@pytest.mark.anyio
async def test_tool_structured_output_list(self):
@ -512,9 +512,9 @@ class TestServerTools:
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("get_numbers", {})
assert result.isError is False
assert result.structuredContent is not None
assert result.structuredContent == {"result": [1, 2, 3, 4, 5]}
assert result.is_error is False
assert result.structured_content is not None
assert result.structured_content == {"result": [1, 2, 3, 4, 5]}
@pytest.mark.anyio
async def test_tool_structured_output_server_side_validation_error(self):
@ -528,8 +528,8 @@ class TestServerTools:
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("get_numbers", {})
assert result.isError is True
assert result.structuredContent is None
assert result.is_error is True
assert result.structured_content is None
assert len(result.content) == 1
assert isinstance(result.content[0], TextContent)
@ -554,17 +554,18 @@ class TestServerTools:
# Check schema
tools = await client.list_tools()
tool = next(t for t in tools.tools if t.name == "get_metadata")
assert tool.outputSchema is not None
assert tool.outputSchema["type"] == "object"
assert tool.output_schema is not None
assert tool.output_schema["type"] == "object"
# dict[str, Any] should have minimal schema
assert (
"additionalProperties" not in tool.outputSchema or tool.outputSchema.get("additionalProperties") is True
"additionalProperties" not in tool.output_schema
or tool.output_schema.get("additionalProperties") is True
)
# Call tool
result = await client.call_tool("get_metadata", {})
assert result.isError is False
assert result.structuredContent is not None
assert result.is_error is False
assert result.structured_content is not None
expected = {
"version": "1.0.0",
"enabled": True,
@ -572,7 +573,7 @@ class TestServerTools:
"tags": ["production", "stable"],
"config": {"nested": {"value": 123}},
}
assert result.structuredContent == expected
assert result.structured_content == expected
@pytest.mark.anyio
async def test_tool_structured_output_dict_str_typed(self):
@ -589,14 +590,14 @@ class TestServerTools:
# Check schema
tools = await client.list_tools()
tool = next(t for t in tools.tools if t.name == "get_settings")
assert tool.outputSchema is not None
assert tool.outputSchema["type"] == "object"
assert tool.outputSchema["additionalProperties"]["type"] == "string"
assert tool.output_schema is not None
assert tool.output_schema["type"] == "object"
assert tool.output_schema["additionalProperties"]["type"] == "string"
# Call tool
result = await client.call_tool("get_settings", {})
assert result.isError is False
assert result.structuredContent == {"theme": "dark", "language": "en", "timezone": "UTC"}
assert result.is_error is False
assert result.structured_content == {"theme": "dark", "language": "en", "timezone": "UTC"}
@pytest.mark.anyio
async def test_remove_tool(self):
@ -656,7 +657,7 @@ class TestServerTools:
# Verify tool works before removal
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("tool_fn", {"x": 1, "y": 2})
assert not result.isError
assert not result.is_error
content = result.content[0]
assert isinstance(content, TextContent)
assert content.text == "3"
@ -667,7 +668,7 @@ class TestServerTools:
# Verify calling removed tool returns an error
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("tool_fn", {"x": 1, "y": 2})
assert result.isError
assert result.is_error
content = result.content[0]
assert isinstance(content, TextContent)
assert "Unknown tool" in content.text
@ -762,7 +763,7 @@ class TestServerResources:
assert resource.description == "get_data returns a string"
assert resource.uri == "function://test"
assert resource.name == "test_get_data"
assert resource.mimeType == "text/plain"
assert resource.mime_type == "text/plain"
class TestServerResourceTemplates:
@ -892,8 +893,8 @@ class TestServerResourceTemplates:
assert len(templates) == 1
template = templates[0]
assert hasattr(template, "mimeType")
assert template.mimeType == "text/csv"
assert hasattr(template, "mime_type")
assert template.mime_type == "text/csv"
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource("resource://bob/csv")
@ -1382,7 +1383,7 @@ class TestServerPrompts:
resource=TextResourceContents(
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
mime_type="text/plain",
),
)
)
@ -1397,7 +1398,7 @@ class TestServerPrompts:
resource = content.resource
assert isinstance(resource, TextResourceContents)
assert resource.text == "File contents"
assert resource.mimeType == "text/plain"
assert resource.mime_type == "text/plain"
@pytest.mark.anyio
async def test_get_unknown_prompt(self):

22
tests/server/fastmcp/test_title.py
View file

@ -26,10 +26,10 @@ async def test_server_name_title_description_version():
# Start server and connect client
async with create_connected_server_and_client_session(mcp._mcp_server) as client:
init_result = await client.initialize()
assert init_result.serverInfo.name == "TestServer"
assert init_result.serverInfo.title == "Test Server Title"
assert init_result.serverInfo.description == "This is a test server description."
assert init_result.serverInfo.version == "1.0"
assert init_result.server_info.name == "TestServer"
assert init_result.server_info.title == "Test Server Title"
assert init_result.server_info.description == "This is a test server description."
assert init_result.server_info.version == "1.0"
@pytest.mark.anyio
@ -184,7 +184,7 @@ async def test_resource_title():
# List resource templates
templates_result = await client.list_resource_templates()
templates = {tpl.uriTemplate: tpl for tpl in templates_result.resourceTemplates}
templates = {tpl.uri_template: tpl for tpl in templates_result.resource_templates}
# Verify dynamic resource template
assert "resource://dynamic/{id}" in templates
@ -203,17 +203,17 @@ async def test_get_display_name_utility():
"""Test the get_display_name utility function."""
# Test tool precedence: title > annotations.title > name
tool_name_only = Tool(name="test_tool", inputSchema={})
tool_name_only = Tool(name="test_tool", input_schema={})
assert get_display_name(tool_name_only) == "test_tool"
tool_with_title = Tool(name="test_tool", title="Test Tool", inputSchema={})
tool_with_title = Tool(name="test_tool", title="Test Tool", input_schema={})
assert get_display_name(tool_with_title) == "Test Tool"
tool_with_annotations = Tool(name="test_tool", inputSchema={}, annotations=ToolAnnotations(title="Annotated Tool"))
tool_with_annotations = Tool(name="test_tool", input_schema={}, annotations=ToolAnnotations(title="Annotated Tool"))
assert get_display_name(tool_with_annotations) == "Annotated Tool"
tool_with_both = Tool(
name="test_tool", title="Primary Title", inputSchema={}, annotations=ToolAnnotations(title="Secondary Title")
name="test_tool", title="Primary Title", input_schema={}, annotations=ToolAnnotations(title="Secondary Title")
)
assert get_display_name(tool_with_both) == "Primary Title"
@ -230,8 +230,8 @@ async def test_get_display_name_utility():
prompt_with_title = Prompt(name="test_prompt", title="Test Prompt")
assert get_display_name(prompt_with_title) == "Test Prompt"
template = ResourceTemplate(uriTemplate="file://{id}", name="test_template")
template = ResourceTemplate(uri_template="file://{id}", name="test_template")
assert get_display_name(template) == "test_template"
template_with_title = ResourceTemplate(uriTemplate="file://{id}", name="test_template", title="Test Template")
template_with_title = ResourceTemplate(uri_template="file://{id}", name="test_template", title="Test Template")
assert get_display_name(template_with_title) == "Test Template"

16
tests/server/fastmcp/test_tool_manager.py
View file

@ -426,8 +426,8 @@ class TestToolAnnotations:
annotations = ToolAnnotations(
title="File Reader",
readOnlyHint=True,
openWorldHint=False,
read_only_hint=True,
open_world_hint=False,
)
manager = ToolManager()
@ -435,8 +435,8 @@ class TestToolAnnotations:
assert tool.annotations is not None
assert tool.annotations.title == "File Reader"
assert tool.annotations.readOnlyHint is True
assert tool.annotations.openWorldHint is False
assert tool.annotations.read_only_hint is True
assert tool.annotations.open_world_hint is False
@pytest.mark.anyio
async def test_tool_annotations_in_fastmcp(self):
@ -444,7 +444,7 @@ class TestToolAnnotations:
app = FastMCP()
@app.tool(annotations=ToolAnnotations(title="Echo Tool", readOnlyHint=True))
@app.tool(annotations=ToolAnnotations(title="Echo Tool", read_only_hint=True))
def echo(message: str) -> str: # pragma: no cover
"""Echo a message back."""
return message
@ -453,7 +453,7 @@ class TestToolAnnotations:
assert len(tools) == 1
assert tools[0].annotations is not None
assert tools[0].annotations.title == "Echo Tool"
assert tools[0].annotations.readOnlyHint is True
assert tools[0].annotations.read_only_hint is True
class TestStructuredOutput:
@ -794,7 +794,7 @@ class TestToolMetadata:
app = FastMCP()
metadata = {"custom": "value"}
annotations = ToolAnnotations(title="Combined Tool", readOnlyHint=True)
annotations = ToolAnnotations(title="Combined Tool", read_only_hint=True)
@app.tool(meta=metadata, annotations=annotations)
def combined_tool(data: str) -> str: # pragma: no cover
@ -806,7 +806,7 @@ class TestToolMetadata:
assert tools[0].meta == metadata
assert tools[0].annotations is not None
assert tools[0].annotations.title == "Combined Tool"
assert tools[0].annotations.readOnlyHint is True
assert tools[0].annotations.read_only_hint is True
class TestRemoveTools:

16
tests/server/fastmcp/test_url_elicitation.py
View file

@ -32,7 +32,7 @@ async def test_url_elicitation_accept():
async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams):
assert params.mode == "url"
assert params.url == "https://example.com/api_key_setup"
assert params.elicitationId == "test-elicitation-001"
assert params.elicitation_id == "test-elicitation-001"
assert params.message == "Please provide your API key to continue."
return ElicitResult(action="accept")
@ -160,9 +160,9 @@ async def test_url_no_content_in_response():
# Verify that this is URL mode
assert params.mode == "url"
assert isinstance(params, types.ElicitRequestURLParams)
# URL params have url and elicitationId, not requestedSchema
# URL params have url and elicitation_id, not requested_schema
assert params.url == "https://example.com/test"
assert params.elicitationId == "test-001"
assert params.elicitation_id == "test-001"
# Return without content - this is correct for URL mode
return ElicitResult(action="accept")
@ -199,8 +199,8 @@ async def test_form_mode_still_works():
# Verify form mode parameters
assert params.mode == "form"
assert isinstance(params, types.ElicitRequestFormParams)
# Form params have requestedSchema, not url/elicitationId
assert params.requestedSchema is not None
# Form params have requested_schema, not url/elicitation_id
assert params.requested_schema is not None
return ElicitResult(action="accept", content={"name": "Alice"})
async with create_connected_server_and_client_session(
@ -341,7 +341,7 @@ async def test_deprecated_elicit_method():
# Use the deprecated elicit() method which should call elicit_form()
result = await ctx.session.elicit(
message="Enter your email",
requestedSchema=EmailSchema.model_json_schema(),
requested_schema=EmailSchema.model_json_schema(),
)
if result.action == "accept" and result.content:
@ -351,7 +351,7 @@ async def test_deprecated_elicit_method():
async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams):
# Verify this is form mode
assert params.mode == "form"
assert params.requestedSchema is not None
assert params.requested_schema is not None
return ElicitResult(action="accept", content={"email": "test@example.com"})
async with create_connected_server_and_client_session(
@ -382,7 +382,7 @@ async def test_ctx_elicit_url_convenience_method():
async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams):
assert params.mode == "url"
assert params.elicitationId == "ctx-test-001"
assert params.elicitation_id == "ctx-test-001"
return ElicitResult(action="accept")
async with create_connected_server_and_client_session(

14
tests/server/fastmcp/test_url_elicitation_error_throw.py
View file

@ -23,7 +23,7 @@ async def test_url_elicitation_error_thrown_from_tool():
mode="url",
message=f"Authorization required to connect to {service_name}",
url=f"https://{service_name}.example.com/oauth/authorize",
elicitationId=f"{service_name}-auth-001",
elicitation_id=f"{service_name}-auth-001",
)
]
)
@ -63,13 +63,13 @@ async def test_url_elicitation_error_from_error():
mode="url",
message="GitHub authorization required",
url="https://github.example.com/oauth",
elicitationId="github-auth",
elicitation_id="github-auth",
),
types.ElicitRequestURLParams(
mode="url",
message="Google Drive authorization required",
url="https://drive.google.com/oauth",
elicitationId="gdrive-auth",
elicitation_id="gdrive-auth",
),
]
)
@ -89,13 +89,13 @@ async def test_url_elicitation_error_from_error():
# Verify the reconstructed error has both elicitations
assert len(url_error.elicitations) == 2
assert url_error.elicitations[0].elicitationId == "github-auth"
assert url_error.elicitations[1].elicitationId == "gdrive-auth"
assert url_error.elicitations[0].elicitation_id == "github-auth"
assert url_error.elicitations[1].elicitation_id == "gdrive-auth"
@pytest.mark.anyio
async def test_normal_exceptions_still_return_error_result():
"""Test that normal exceptions still return CallToolResult with isError=True."""
"""Test that normal exceptions still return CallToolResult with is_error=True."""
mcp = FastMCP(name="NormalErrorServer")
@mcp.tool(description="A tool that raises a normal exception")
@ -107,7 +107,7 @@ async def test_normal_exceptions_still_return_error_result():
# Normal exceptions should be returned as error results, not McpError
result = await client_session.call_tool("failing_tool", {})
assert result.isError is True
assert result.is_error is True
assert len(result.content) == 1
assert isinstance(result.content[0], types.TextContent)
assert "Something went wrong" in result.content[0].text

4
tests/server/lowlevel/test_server_listing.py
View file

@ -80,7 +80,7 @@ async def test_list_tools_basic() -> None:
Tool(
name="tool1",
description="First tool",
inputSchema={
input_schema={
"type": "object",
"properties": {
"message": {"type": "string"},
@ -91,7 +91,7 @@ async def test_list_tools_basic() -> None:
Tool(
name="tool2",
description="Second tool",
inputSchema={
input_schema={
"type": "object",
"properties": {
"count": {"type": "number"},

6
tests/server/lowlevel/test_server_pagination.py
View file

@ -25,7 +25,7 @@ async def test_list_prompts_pagination() -> None:
async def handle_list_prompts(request: ListPromptsRequest) -> ListPromptsResult:
nonlocal received_request
received_request = request
return ListPromptsResult(prompts=[], nextCursor="next")
return ListPromptsResult(prompts=[], next_cursor="next")
handler = server.request_handlers[ListPromptsRequest]
@ -57,7 +57,7 @@ async def test_list_resources_pagination() -> None:
async def handle_list_resources(request: ListResourcesRequest) -> ListResourcesResult:
nonlocal received_request
received_request = request
return ListResourcesResult(resources=[], nextCursor="next")
return ListResourcesResult(resources=[], next_cursor="next")
handler = server.request_handlers[ListResourcesRequest]
@ -91,7 +91,7 @@ async def test_list_tools_pagination() -> None:
async def handle_list_tools(request: ListToolsRequest) -> ListToolsResult:
nonlocal received_request
received_request = request
return ListToolsResult(tools=[], nextCursor="next")
return ListToolsResult(tools=[], next_cursor="next")
handler = server.request_handlers[ListToolsRequest]

4
tests/server/test_cancel_handling.py
View file

@ -38,7 +38,7 @@ async def test_server_remains_functional_after_cancel():
Tool(
name="test_tool",
description="Tool for testing",
inputSchema={},
input_schema={},
)
]
@ -83,7 +83,7 @@ async def test_server_remains_functional_after_cancel():
ClientNotification(
CancelledNotification(
params=CancelledNotificationParams(
requestId=first_request_id,
request_id=first_request_id,
reason="Testing server recovery",
),
)

16
tests/server/test_completion_with_context.py
View file

@ -36,7 +36,7 @@ async def test_completion_handler_receives_context():
received_args["context"] = context
# Return test completion
return Completion(values=["test-completion"], total=1, hasMore=False)
return Completion(values=["test-completion"], total=1, has_more=False)
async with create_connected_server_and_client_session(server) as client:
# Test with context
@ -68,7 +68,7 @@ async def test_completion_backward_compatibility():
nonlocal context_was_none
context_was_none = context is None
return Completion(values=["no-context-completion"], total=1, hasMore=False)
return Completion(values=["no-context-completion"], total=1, has_more=False)
async with create_connected_server_and_client_session(server) as client:
# Test without context
@ -97,17 +97,17 @@ async def test_dependent_completion_scenario():
if ref.uri == "db://{database}/{table}":
if argument.name == "database":
# Complete database names
return Completion(values=["users_db", "products_db", "analytics_db"], total=3, hasMore=False)
return Completion(values=["users_db", "products_db", "analytics_db"], total=3, has_more=False)
elif argument.name == "table":
# Complete table names based on selected database
if context and context.arguments:
db = context.arguments.get("database")
if db == "users_db":
return Completion(values=["users", "sessions", "permissions"], total=3, hasMore=False)
return Completion(values=["users", "sessions", "permissions"], total=3, has_more=False)
elif db == "products_db": # pragma: no cover
return Completion(values=["products", "categories", "inventory"], total=3, hasMore=False)
return Completion(values=["products", "categories", "inventory"], total=3, has_more=False)
return Completion(values=[], total=0, hasMore=False) # pragma: no cover
return Completion(values=[], total=0, has_more=False) # pragma: no cover
async with create_connected_server_and_client_session(server) as client:
# First, complete database
@ -156,9 +156,9 @@ async def test_completion_error_on_missing_context():
# Normal completion if context is provided
db = context.arguments.get("database")
if db == "test_db": # pragma: no cover
return Completion(values=["users", "orders", "products"], total=3, hasMore=False)
return Completion(values=["users", "orders", "products"], total=3, has_more=False)
return Completion(values=[], total=0, hasMore=False) # pragma: no cover
return Completion(values=[], total=0, has_more=False) # pragma: no cover
async with create_connected_server_and_client_session(server) as client:
# Try to complete table without database context - should raise error

8
tests/server/test_lifespan.py
View file

@ -76,9 +76,9 @@ async def test_lowlevel_server_lifespan():
# Initialize the server
params = InitializeRequestParams(
protocolVersion="2024-11-05",
protocol_version="2024-11-05",
capabilities=ClientCapabilities(),
clientInfo=Implementation(name="test-client", version="0.1.0"),
client_info=Implementation(name="test-client", version="0.1.0"),
)
await send_stream1.send(
SessionMessage(
@ -182,9 +182,9 @@ async def test_fastmcp_server_lifespan():
# Initialize the server
params = InitializeRequestParams(
protocolVersion="2024-11-05",
protocol_version="2024-11-05",
capabilities=ClientCapabilities(),
clientInfo=Implementation(name="test-client", version="0.1.0"),
client_info=Implementation(name="test-client", version="0.1.0"),
)
await send_stream1.send(
SessionMessage(

20
tests/server/test_lowlevel_input_validation.py
View file

@ -103,7 +103,7 @@ def create_add_tool() -> Tool:
return Tool(
name="add",
description="Add two numbers",
inputSchema={
input_schema={
"type": "object",
"properties": {
"a": {"type": "number"},
@ -133,7 +133,7 @@ async def test_valid_tool_call():
# Verify results
assert result is not None
assert not result.isError
assert not result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert isinstance(result.content[0], TextContent)
@ -155,7 +155,7 @@ async def test_invalid_tool_call_missing_required():
# Verify results
assert result is not None
assert result.isError
assert result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert isinstance(result.content[0], TextContent)
@ -178,7 +178,7 @@ async def test_invalid_tool_call_wrong_type():
# Verify results
assert result is not None
assert result.isError
assert result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert isinstance(result.content[0], TextContent)
@ -193,7 +193,7 @@ async def test_cache_refresh_on_missing_tool():
Tool(
name="multiply",
description="Multiply two numbers",
inputSchema={
input_schema={
"type": "object",
"properties": {
"x": {"type": "number"},
@ -220,7 +220,7 @@ async def test_cache_refresh_on_missing_tool():
# Verify results - should work because cache will be refreshed
assert result is not None
assert not result.isError
assert not result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert isinstance(result.content[0], TextContent)
@ -234,7 +234,7 @@ async def test_enum_constraint_validation():
Tool(
name="greet",
description="Greet someone",
inputSchema={
input_schema={
"type": "object",
"properties": {
"name": {"type": "string"},
@ -256,7 +256,7 @@ async def test_enum_constraint_validation():
# Verify results
assert result is not None
assert result.isError
assert result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert isinstance(result.content[0], TextContent)
@ -271,7 +271,7 @@ async def test_tool_not_in_list_logs_warning(caplog: pytest.LogCaptureFixture):
Tool(
name="add",
description="Add two numbers",
inputSchema={
input_schema={
"type": "object",
"properties": {
"a": {"type": "number"},
@ -300,7 +300,7 @@ async def test_tool_not_in_list_logs_warning(caplog: pytest.LogCaptureFixture):
# Verify results - should succeed because validation is skipped for unknown tools
assert result is not None
assert not result.isError
assert not result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert isinstance(result.content[0], TextContent)

60
tests/server/test_lowlevel_output_validation.py
View file

@ -106,7 +106,7 @@ async def test_content_only_without_output_schema():
Tool(
name="echo",
description="Echo a message",
inputSchema={
input_schema={
"type": "object",
"properties": {
"message": {"type": "string"},
@ -130,12 +130,12 @@ async def test_content_only_without_output_schema():
# Verify results
assert result is not None
assert not result.isError
assert not result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Echo: Hello"
assert result.structuredContent is None
assert result.structured_content is None
@pytest.mark.anyio
@ -145,7 +145,7 @@ async def test_dict_only_without_output_schema():
Tool(
name="get_info",
description="Get structured information",
inputSchema={
input_schema={
"type": "object",
"properties": {},
},
@ -166,13 +166,13 @@ async def test_dict_only_without_output_schema():
# Verify results
assert result is not None
assert not result.isError
assert not result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert isinstance(result.content[0], TextContent)
# Check that the content is the JSON serialization
assert json.loads(result.content[0].text) == {"status": "ok", "data": {"value": 42}}
assert result.structuredContent == {"status": "ok", "data": {"value": 42}}
assert result.structured_content == {"status": "ok", "data": {"value": 42}}
@pytest.mark.anyio
@ -182,7 +182,7 @@ async def test_both_content_and_dict_without_output_schema():
Tool(
name="process",
description="Process data",
inputSchema={
input_schema={
"type": "object",
"properties": {},
},
@ -205,12 +205,12 @@ async def test_both_content_and_dict_without_output_schema():
# Verify results
assert result is not None
assert not result.isError
assert not result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert isinstance(result.content[0], TextContent)
assert result.content[0].text == "Processing complete"
assert result.structuredContent == {"result": "success", "count": 10}
assert result.structured_content == {"result": "success", "count": 10}
@pytest.mark.anyio
@ -220,11 +220,11 @@ async def test_content_only_with_output_schema_error():
Tool(
name="structured_tool",
description="Tool expecting structured output",
inputSchema={
input_schema={
"type": "object",
"properties": {},
},
outputSchema={
output_schema={
"type": "object",
"properties": {
"result": {"type": "string"},
@ -245,7 +245,7 @@ async def test_content_only_with_output_schema_error():
# Verify error
assert result is not None
assert result.isError
assert result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert isinstance(result.content[0], TextContent)
@ -259,7 +259,7 @@ async def test_valid_dict_with_output_schema():
Tool(
name="calc",
description="Calculate result",
inputSchema={
input_schema={
"type": "object",
"properties": {
"x": {"type": "number"},
@ -267,7 +267,7 @@ async def test_valid_dict_with_output_schema():
},
"required": ["x", "y"],
},
outputSchema={
output_schema={
"type": "object",
"properties": {
"sum": {"type": "number"},
@ -293,12 +293,12 @@ async def test_valid_dict_with_output_schema():
# Verify results
assert result is not None
assert not result.isError
assert not result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
# Check JSON serialization
assert json.loads(result.content[0].text) == {"sum": 7, "product": 12}
assert result.structuredContent == {"sum": 7, "product": 12}
assert result.structured_content == {"sum": 7, "product": 12}
@pytest.mark.anyio
@ -308,11 +308,11 @@ async def test_invalid_dict_with_output_schema():
Tool(
name="user_info",
description="Get user information",
inputSchema={
input_schema={
"type": "object",
"properties": {},
},
outputSchema={
output_schema={
"type": "object",
"properties": {
"name": {"type": "string"},
@ -337,7 +337,7 @@ async def test_invalid_dict_with_output_schema():
# Verify error
assert result is not None
assert result.isError
assert result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert isinstance(result.content[0], TextContent)
@ -352,14 +352,14 @@ async def test_both_content_and_valid_dict_with_output_schema():
Tool(
name="analyze",
description="Analyze data",
inputSchema={
input_schema={
"type": "object",
"properties": {
"text": {"type": "string"},
},
"required": ["text"],
},
outputSchema={
output_schema={
"type": "object",
"properties": {
"sentiment": {"type": "string", "enum": ["positive", "negative", "neutral"]},
@ -385,11 +385,11 @@ async def test_both_content_and_valid_dict_with_output_schema():
# Verify results
assert result is not None
assert not result.isError
assert not result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert result.content[0].text == "Analysis of: Great job!"
assert result.structuredContent == {"sentiment": "positive", "confidence": 0.95}
assert result.structured_content == {"sentiment": "positive", "confidence": 0.95}
@pytest.mark.anyio
@ -399,7 +399,7 @@ async def test_tool_call_result():
Tool(
name="get_info",
description="Get structured information",
inputSchema={
input_schema={
"type": "object",
"properties": {},
},
@ -411,7 +411,7 @@ async def test_tool_call_result():
if name == "get_info":
return CallToolResult(
content=[TextContent(type="text", text="Results calculated")],
structuredContent={"status": "ok", "data": {"value": 42}},
structured_content={"status": "ok", "data": {"value": 42}},
_meta={"some": "metadata"},
)
else: # pragma: no cover
@ -424,12 +424,12 @@ async def test_tool_call_result():
# Verify results
assert result is not None
assert not result.isError
assert not result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert result.content[0].text == "Results calculated"
assert isinstance(result.content[0], TextContent)
assert result.structuredContent == {"status": "ok", "data": {"value": 42}}
assert result.structured_content == {"status": "ok", "data": {"value": 42}}
assert result.meta == {"some": "metadata"}
@ -440,11 +440,11 @@ async def test_output_schema_type_validation():
Tool(
name="stats",
description="Get statistics",
inputSchema={
input_schema={
"type": "object",
"properties": {},
},
outputSchema={
output_schema={
"type": "object",
"properties": {
"count": {"type": "integer"},
@ -470,7 +470,7 @@ async def test_output_schema_type_validation():
# Verify error
assert result is not None
assert result.isError
assert result.is_error
assert len(result.content) == 1
assert result.content[0].type == "text"
assert "Output validation error:" in result.content[0].text

6
tests/server/test_lowlevel_tool_annotations.py
View file

@ -25,7 +25,7 @@ async def test_lowlevel_server_tool_annotations():
Tool(
name="echo",
description="Echo a message back",
inputSchema={
input_schema={
"type": "object",
"properties": {
"message": {"type": "string"},
@ -34,7 +34,7 @@ async def test_lowlevel_server_tool_annotations():
},
annotations=ToolAnnotations(
title="Echo Tool",
readOnlyHint=True,
read_only_hint=True,
),
)
]
@ -98,4 +98,4 @@ async def test_lowlevel_server_tool_annotations():
assert tools_result.tools[0].name == "echo"
assert tools_result.tools[0].annotations is not None
assert tools_result.tools[0].annotations.title == "Echo Tool"
assert tools_result.tools[0].annotations.readOnlyHint is True
assert tools_result.tools[0].annotations.read_only_hint is True

6
tests/server/test_read_resource.py
View file

@ -45,7 +45,7 @@ async def test_read_resource_text(temp_file: Path):
content = result.root.contents[0]
assert isinstance(content, types.TextResourceContents)
assert content.text == "Hello World"
assert content.mimeType == "text/plain"
assert content.mime_type == "text/plain"
@pytest.mark.anyio
@ -71,7 +71,7 @@ async def test_read_resource_binary(temp_file: Path):
content = result.root.contents[0]
assert isinstance(content, types.BlobResourceContents)
assert content.mimeType == "application/octet-stream"
assert content.mime_type == "application/octet-stream"
@pytest.mark.anyio
@ -103,4 +103,4 @@ async def test_read_resource_default_mime(temp_file: Path):
content = result.root.contents[0]
assert isinstance(content, types.TextResourceContents)
assert content.text == "Hello World"
assert content.mimeType == "text/plain"
assert content.mime_type == "text/plain"

32
tests/server/test_session.py
View file

@ -101,7 +101,7 @@ async def test_server_capabilities():
return []
caps = server.get_capabilities(notification_options, experimental_capabilities)
assert caps.prompts == PromptsCapability(listChanged=False)
assert caps.prompts == PromptsCapability(list_changed=False)
assert caps.resources is None
assert caps.completions is None
@ -111,8 +111,8 @@ async def test_server_capabilities():
return []
caps = server.get_capabilities(notification_options, experimental_capabilities)
assert caps.prompts == PromptsCapability(listChanged=False)
assert caps.resources == ResourcesCapability(subscribe=False, listChanged=False)
assert caps.prompts == PromptsCapability(list_changed=False)
assert caps.resources == ResourcesCapability(subscribe=False, list_changed=False)
assert caps.completions is None
# Add a complete handler
@ -127,8 +127,8 @@ async def test_server_capabilities():
)
caps = server.get_capabilities(notification_options, experimental_capabilities)
assert caps.prompts == PromptsCapability(listChanged=False)
assert caps.resources == ResourcesCapability(subscribe=False, listChanged=False)
assert caps.prompts == PromptsCapability(list_changed=False)
assert caps.resources == ResourcesCapability(subscribe=False, list_changed=False)
assert caps.completions == CompletionsCapability()
@ -175,9 +175,9 @@ async def test_server_session_initialize_with_older_protocol_version():
id=1,
method="initialize",
params=types.InitializeRequestParams(
protocolVersion="2024-11-05",
protocol_version="2024-11-05",
capabilities=types.ClientCapabilities(),
clientInfo=types.Implementation(name="test-client", version="1.0.0"),
client_info=types.Implementation(name="test-client", version="1.0.0"),
).model_dump(by_alias=True, mode="json", exclude_none=True),
)
)
@ -191,7 +191,7 @@ async def test_server_session_initialize_with_older_protocol_version():
init_result = types.InitializeResult.model_validate(result_data)
# Check that the server responded with the requested protocol version
received_protocol_version = init_result.protocolVersion
received_protocol_version = init_result.protocol_version
assert received_protocol_version == "2024-11-05"
# Send initialized notification
@ -312,17 +312,17 @@ async def test_create_message_tool_result_validation():
) as session:
# Set up client params with sampling.tools capability for the test
session._client_params = types.InitializeRequestParams(
protocolVersion=types.LATEST_PROTOCOL_VERSION,
protocol_version=types.LATEST_PROTOCOL_VERSION,
capabilities=types.ClientCapabilities(
sampling=types.SamplingCapability(tools=types.SamplingToolsCapability())
),
clientInfo=types.Implementation(name="test", version="1.0"),
client_info=types.Implementation(name="test", version="1.0"),
)
tool = types.Tool(name="test_tool", inputSchema={"type": "object"})
tool = types.Tool(name="test_tool", input_schema={"type": "object"})
text = types.TextContent(type="text", text="hello")
tool_use = types.ToolUseContent(type="tool_use", id="call_1", name="test_tool", input={})
tool_result = types.ToolResultContent(type="tool_result", toolUseId="call_1", content=[])
tool_result = types.ToolResultContent(type="tool_result", tool_use_id="call_1", content=[])
# Case 1: tool_result mixed with other content
with pytest.raises(ValueError, match="only tool_result content"):
@ -363,7 +363,7 @@ async def test_create_message_tool_result_validation():
types.SamplingMessage(role="assistant", content=tool_use),
types.SamplingMessage(
role="user",
content=types.ToolResultContent(type="tool_result", toolUseId="wrong_id", content=[]),
content=types.ToolResultContent(type="tool_result", tool_use_id="wrong_id", content=[]),
),
],
max_tokens=100,
@ -438,12 +438,12 @@ async def test_create_message_without_tools_capability():
) as session:
# Set up client params WITHOUT sampling.tools capability
session._client_params = types.InitializeRequestParams(
protocolVersion=types.LATEST_PROTOCOL_VERSION,
protocol_version=types.LATEST_PROTOCOL_VERSION,
capabilities=types.ClientCapabilities(sampling=types.SamplingCapability()),
clientInfo=types.Implementation(name="test", version="1.0"),
client_info=types.Implementation(name="test", version="1.0"),
)
tool = types.Tool(name="test_tool", inputSchema={"type": "object"})
tool = types.Tool(name="test_tool", input_schema={"type": "object"})
text = types.TextContent(type="text", text="hello")
# Should raise McpError when tools are provided but client lacks capability

Some files were not shown because too many files have changed in this diff Show more