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

test: convert test_integration.py to in-memory transport (fix flaky) (#2277)

Browse source
This commit is contained in:
Max Isbey 2026-03-18 15:25:11 +00:00 • committed by GitHub
commit 7826ade12b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 231 additions and 515 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/mcp/server/session.py
View file

@ -474,7 +474,7 @@ class ServerSession(
related_request_id, related_request_id,
) )
async def send_resource_list_changed(self) -> None: # pragma: no cover async def send_resource_list_changed(self) -> None:
"""Send a resource list changed notification.""" """Send a resource list changed notification."""
await self.send_notification(types.ResourceListChangedNotification()) await self.send_notification(types.ResourceListChangedNotification())

666
tests/server/mcpserver/test_integration.py
View file

@ -1,7 +1,7 @@
"""Integration tests for MCPServer server functionality. """Integration tests for MCPServer server functionality.
These tests validate the proper functioning of MCPServer features using focused, These tests validate the proper functioning of MCPServer features using focused,
single-feature servers across different transports (SSE and StreamableHTTP). single-feature example servers over an in-memory transport.
""" """
# TODO(Marcelo): The `examples` package is not being imported as package. We need to solve this. # TODO(Marcelo): The `examples` package is not being imported as package. We need to solve this.
# pyright: reportUnknownMemberType=false # pyright: reportUnknownMemberType=false
@ -10,12 +10,8 @@ single-feature servers across different transports (SSE and StreamableHTTP).
# pyright: reportUnknownArgumentType=false # pyright: reportUnknownArgumentType=false
import json import json
import multiprocessing
import socket
from collections.abc import Generator
import pytest import pytest
import uvicorn
from inline_snapshot import snapshot from inline_snapshot import snapshot
from examples.snippets.servers import ( from examples.snippets.servers import (
@ -30,9 +26,8 @@ from examples.snippets.servers import (
structured_output, structured_output,
tool_progress, tool_progress,
) )
from mcp.client import Client
from mcp.client.session import ClientSession from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
from mcp.client.streamable_http import streamable_http_client
from mcp.shared._context import RequestContext from mcp.shared._context import RequestContext
from mcp.shared.session import RequestResponder from mcp.shared.session import RequestResponder
from mcp.types import ( from mcp.types import (
@ -42,7 +37,6 @@ from mcp.types import (
ElicitRequestParams, ElicitRequestParams,
ElicitResult, ElicitResult,
GetPromptResult, GetPromptResult,
InitializeResult,
LoggingMessageNotification, LoggingMessageNotification,
LoggingMessageNotificationParams, LoggingMessageNotificationParams,
NotificationParams, NotificationParams,
@ -58,7 +52,8 @@ from mcp.types import (
TextResourceContents, TextResourceContents,
ToolListChangedNotification, ToolListChangedNotification,
) )
from tests.test_helpers import wait_for_server
pytestmark = pytest.mark.anyio
class NotificationCollector: class NotificationCollector:
@ -85,105 +80,6 @@ class NotificationCollector:
self.tool_notifications.append(message.params) self.tool_notifications.append(message.params)
# Common fixtures
@pytest.fixture
def server_port() -> int:
"""Get a free port for testing."""
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
@pytest.fixture
def server_url(server_port: int) -> str:
"""Get the server URL for testing."""
return f"http://127.0.0.1:{server_port}"
def run_server_with_transport(module_name: str, port: int, transport: str) -> None: # pragma: no cover
"""Run server with specified transport."""
# Get the MCP instance based on module name
if module_name == "basic_tool":
mcp = basic_tool.mcp
elif module_name == "basic_resource":
mcp = basic_resource.mcp
elif module_name == "basic_prompt":
mcp = basic_prompt.mcp
elif module_name == "tool_progress":
mcp = tool_progress.mcp
elif module_name == "sampling":
mcp = sampling.mcp
elif module_name == "elicitation":
mcp = elicitation.mcp
elif module_name == "completion":
mcp = completion.mcp
elif module_name == "notifications":
mcp = notifications.mcp
elif module_name == "mcpserver_quickstart":
mcp = mcpserver_quickstart.mcp
elif module_name == "structured_output":
mcp = structured_output.mcp
else:
raise ImportError(f"Unknown module: {module_name}")
# Create app based on transport type
if transport == "sse":
app = mcp.sse_app()
elif transport == "streamable-http":
app = mcp.streamable_http_app()
else:
raise ValueError(f"Invalid transport for test server: {transport}")
server = uvicorn.Server(config=uvicorn.Config(app=app, host="127.0.0.1", port=port, log_level="error"))
print(f"Starting {transport} server on port {port}")
server.run()
@pytest.fixture
def server_transport(request: pytest.FixtureRequest, server_port: int) -> Generator[str, None, None]:
"""Start server in a separate process with specified MCP instance and transport.
Args:
request: pytest request with param tuple of (module_name, transport)
server_port: Port to run the server on
Yields:
str: The transport type ('sse' or 'streamable_http')
"""
module_name, transport = request.param
proc = multiprocessing.Process(
target=run_server_with_transport,
args=(module_name, server_port, transport),
daemon=True,
)
proc.start()
# Wait for server to be ready
wait_for_server(server_port)
yield transport
proc.kill()
proc.join(timeout=2)
if proc.is_alive(): # pragma: no cover
print("Server process failed to terminate")
# Helper function to create client based on transport
def create_client_for_transport(transport: str, server_url: str):
"""Create the appropriate client context manager based on transport type."""
if transport == "sse":
endpoint = f"{server_url}/sse"
return sse_client(endpoint)
elif transport == "streamable-http":
endpoint = f"{server_url}/mcp"
return streamable_http_client(endpoint)
else: # pragma: no cover
raise ValueError(f"Invalid transport: {transport}")
# Callback functions for testing
async def sampling_callback( async def sampling_callback(
context: RequestContext[ClientSession], params: CreateMessageRequestParams context: RequestContext[ClientSession], params: CreateMessageRequestParams
) -> CreateMessageResult: ) -> CreateMessageResult:
@ -210,147 +106,85 @@ async def elicitation_callback(context: RequestContext[ClientSession], params: E
return ElicitResult(action="decline") return ElicitResult(action="decline")
# Test basic tools async def test_basic_tools() -> None:
@pytest.mark.anyio
@pytest.mark.parametrize(
"server_transport",
[
("basic_tool", "sse"),
("basic_tool", "streamable-http"),
],
indirect=True,
)
async def test_basic_tools(server_transport: str, server_url: str) -> None:
"""Test basic tool functionality.""" """Test basic tool functionality."""
transport = server_transport async with Client(basic_tool.mcp) as client:
client_cm = create_client_for_transport(transport, server_url) assert client.server_capabilities is not None
assert client.server_capabilities.tools is not None
async with client_cm as (read_stream, write_stream): # Test sum tool
async with ClientSession(read_stream, write_stream) as session: tool_result = await client.call_tool("sum", {"a": 5, "b": 3})
# Test initialization assert len(tool_result.content) == 1
result = await session.initialize() assert isinstance(tool_result.content[0], TextContent)
assert isinstance(result, InitializeResult) assert tool_result.content[0].text == "8"
assert result.server_info.name == "Tool Example"
assert result.capabilities.tools is not None
# Test sum tool # Test weather tool
tool_result = await session.call_tool("sum", {"a": 5, "b": 3}) weather_result = await client.call_tool("get_weather", {"city": "London"})
assert len(tool_result.content) == 1 assert len(weather_result.content) == 1
assert isinstance(tool_result.content[0], TextContent) assert isinstance(weather_result.content[0], TextContent)
assert tool_result.content[0].text == "8" assert "Weather in London: 22degreesC" in weather_result.content[0].text
# Test weather tool
weather_result = await session.call_tool("get_weather", {"city": "London"})
assert len(weather_result.content) == 1
assert isinstance(weather_result.content[0], TextContent)
assert "Weather in London: 22degreesC" in weather_result.content[0].text
# Test resources async def test_basic_resources() -> None:
@pytest.mark.anyio
@pytest.mark.parametrize(
"server_transport",
[
("basic_resource", "sse"),
("basic_resource", "streamable-http"),
],
indirect=True,
)
async def test_basic_resources(server_transport: str, server_url: str) -> None:
"""Test basic resource functionality.""" """Test basic resource functionality."""
transport = server_transport async with Client(basic_resource.mcp) as client:
client_cm = create_client_for_transport(transport, server_url) assert client.server_capabilities is not None
assert client.server_capabilities.resources is not None
async with client_cm as (read_stream, write_stream): # Test document resource
async with ClientSession(read_stream, write_stream) as session: doc_content = await client.read_resource("file://documents/readme")
# Test initialization assert isinstance(doc_content, ReadResourceResult)
result = await session.initialize() assert len(doc_content.contents) == 1
assert isinstance(result, InitializeResult) assert isinstance(doc_content.contents[0], TextResourceContents)
assert result.server_info.name == "Resource Example" assert "Content of readme" in doc_content.contents[0].text
assert result.capabilities.resources is not None
# Test document resource # Test settings resource
doc_content = await session.read_resource("file://documents/readme") settings_content = await client.read_resource("config://settings")
assert isinstance(doc_content, ReadResourceResult) assert isinstance(settings_content, ReadResourceResult)
assert len(doc_content.contents) == 1 assert len(settings_content.contents) == 1
assert isinstance(doc_content.contents[0], TextResourceContents) assert isinstance(settings_content.contents[0], TextResourceContents)
assert "Content of readme" in doc_content.contents[0].text settings_json = json.loads(settings_content.contents[0].text)
assert settings_json["theme"] == "dark"
# Test settings resource assert settings_json["language"] == "en"
settings_content = await session.read_resource("config://settings")
assert isinstance(settings_content, ReadResourceResult)
assert len(settings_content.contents) == 1
assert isinstance(settings_content.contents[0], TextResourceContents)
settings_json = json.loads(settings_content.contents[0].text)
assert settings_json["theme"] == "dark"
assert settings_json["language"] == "en"
# Test prompts async def test_basic_prompts() -> None:
@pytest.mark.anyio
@pytest.mark.parametrize(
"server_transport",
[
("basic_prompt", "sse"),
("basic_prompt", "streamable-http"),
],
indirect=True,
)
async def test_basic_prompts(server_transport: str, server_url: str) -> None:
"""Test basic prompt functionality.""" """Test basic prompt functionality."""
transport = server_transport async with Client(basic_prompt.mcp) as client:
client_cm = create_client_for_transport(transport, server_url) assert client.server_capabilities is not None
assert client.server_capabilities.prompts is not None
async with client_cm as (read_stream, write_stream): # Test review_code prompt
async with ClientSession(read_stream, write_stream) as session: prompts = await client.list_prompts()
# Test initialization review_prompt = next((p for p in prompts.prompts if p.name == "review_code"), None)
result = await session.initialize() assert review_prompt is not None
assert isinstance(result, InitializeResult)
assert result.server_info.name == "Prompt Example"
assert result.capabilities.prompts is not None
# Test review_code prompt prompt_result = await client.get_prompt("review_code", {"code": "def hello():\n print('Hello')"})
prompts = await session.list_prompts() assert isinstance(prompt_result, GetPromptResult)
review_prompt = next((p for p in prompts.prompts if p.name == "review_code"), None) assert len(prompt_result.messages) == 1
assert review_prompt is not None assert isinstance(prompt_result.messages[0].content, TextContent)
assert "Please review this code:" in prompt_result.messages[0].content.text
assert "def hello():" in prompt_result.messages[0].content.text
prompt_result = await session.get_prompt("review_code", {"code": "def hello():\n print('Hello')"}) # Test debug_error prompt
assert isinstance(prompt_result, GetPromptResult) debug_result = await client.get_prompt(
assert len(prompt_result.messages) == 1 "debug_error", {"error": "TypeError: 'NoneType' object is not subscriptable"}
assert isinstance(prompt_result.messages[0].content, TextContent) )
assert "Please review this code:" in prompt_result.messages[0].content.text assert isinstance(debug_result, GetPromptResult)
assert "def hello():" in prompt_result.messages[0].content.text assert len(debug_result.messages) == 3
assert debug_result.messages[0].role == "user"
# Test debug_error prompt assert isinstance(debug_result.messages[0].content, TextContent)
debug_result = await session.get_prompt( assert "I'm seeing this error:" in debug_result.messages[0].content.text
"debug_error", {"error": "TypeError: 'NoneType' object is not subscriptable"} assert debug_result.messages[1].role == "user"
) assert isinstance(debug_result.messages[1].content, TextContent)
assert isinstance(debug_result, GetPromptResult) assert "TypeError" in debug_result.messages[1].content.text
assert len(debug_result.messages) == 3 assert debug_result.messages[2].role == "assistant"
assert debug_result.messages[0].role == "user" assert isinstance(debug_result.messages[2].content, TextContent)
assert isinstance(debug_result.messages[0].content, TextContent) assert "I'll help debug that" in debug_result.messages[2].content.text
assert "I'm seeing this error:" in debug_result.messages[0].content.text
assert debug_result.messages[1].role == "user"
assert isinstance(debug_result.messages[1].content, TextContent)
assert "TypeError" in debug_result.messages[1].content.text
assert debug_result.messages[2].role == "assistant"
assert isinstance(debug_result.messages[2].content, TextContent)
assert "I'll help debug that" in debug_result.messages[2].content.text
# Test progress reporting async def test_tool_progress() -> None:
@pytest.mark.anyio
@pytest.mark.parametrize(
"server_transport",
[
("tool_progress", "sse"),
("tool_progress", "streamable-http"),
],
indirect=True,
)
async def test_tool_progress(server_transport: str, server_url: str) -> None:
"""Test tool progress reporting.""" """Test tool progress reporting."""
transport = server_transport
collector = NotificationCollector() collector = NotificationCollector()
async def message_handler(message: RequestResponder[ServerRequest, ClientResult] | ServerNotification | Exception): async def message_handler(message: RequestResponder[ServerRequest, ClientResult] | ServerNotification | Exception):
@ -358,134 +192,79 @@ async def test_tool_progress(server_transport: str, server_url: str) -> None:
if isinstance(message, Exception): # pragma: no cover if isinstance(message, Exception): # pragma: no cover
raise message raise message
client_cm = create_client_for_transport(transport, server_url) async with Client(tool_progress.mcp, message_handler=message_handler) as client:
# Test progress callback
progress_updates = []
async with client_cm as (read_stream, write_stream): async def progress_callback(progress: float, total: float | None, message: str | None) -> None:
async with ClientSession(read_stream, write_stream, message_handler=message_handler) as session: progress_updates.append((progress, total, message))
# Test initialization
result = await session.initialize()
assert isinstance(result, InitializeResult)
assert result.server_info.name == "Progress Example"
# Test progress callback # Call tool with progress
progress_updates = [] steps = 3
tool_result = await client.call_tool(
"long_running_task",
{"task_name": "Test Task", "steps": steps},
progress_callback=progress_callback,
)
assert tool_result.content == snapshot([TextContent(text="Task 'Test Task' completed")])
async def progress_callback(progress: float, total: float | None, message: str | None) -> None: # Verify progress updates
progress_updates.append((progress, total, message)) assert len(progress_updates) == steps
for i, (progress, total, message) in enumerate(progress_updates):
expected_progress = (i + 1) / steps
assert abs(progress - expected_progress) < 0.01
assert total == 1.0
assert f"Step {i + 1}/{steps}" in message
# Call tool with progress # Verify log messages
steps = 3 assert len(collector.log_messages) > 0
tool_result = await session.call_tool(
"long_running_task",
{"task_name": "Test Task", "steps": steps},
progress_callback=progress_callback,
)
assert tool_result.content == snapshot([TextContent(text="Task 'Test Task' completed")])
# Verify progress updates
assert len(progress_updates) == steps
for i, (progress, total, message) in enumerate(progress_updates):
expected_progress = (i + 1) / steps
assert abs(progress - expected_progress) < 0.01
assert total == 1.0
assert f"Step {i + 1}/{steps}" in message
# Verify log messages
assert len(collector.log_messages) > 0
# Test sampling async def test_sampling() -> None:
@pytest.mark.anyio
@pytest.mark.parametrize(
"server_transport",
[
("sampling", "sse"),
("sampling", "streamable-http"),
],
indirect=True,
)
async def test_sampling(server_transport: str, server_url: str) -> None:
"""Test sampling (LLM interaction) functionality.""" """Test sampling (LLM interaction) functionality."""
transport = server_transport async with Client(sampling.mcp, sampling_callback=sampling_callback) as client:
client_cm = create_client_for_transport(transport, server_url) assert client.server_capabilities is not None
assert client.server_capabilities.tools is not None
async with client_cm as (read_stream, write_stream): # Test sampling tool
async with ClientSession(read_stream, write_stream, sampling_callback=sampling_callback) as session: sampling_result = await client.call_tool("generate_poem", {"topic": "nature"})
# Test initialization assert len(sampling_result.content) == 1
result = await session.initialize() assert isinstance(sampling_result.content[0], TextContent)
assert isinstance(result, InitializeResult) assert "This is a simulated LLM response" in sampling_result.content[0].text
assert result.server_info.name == "Sampling Example"
assert result.capabilities.tools is not None
# Test sampling tool
sampling_result = await session.call_tool("generate_poem", {"topic": "nature"})
assert len(sampling_result.content) == 1
assert isinstance(sampling_result.content[0], TextContent)
assert "This is a simulated LLM response" in sampling_result.content[0].text
# Test elicitation async def test_elicitation() -> None:
@pytest.mark.anyio
@pytest.mark.parametrize(
"server_transport",
[
("elicitation", "sse"),
("elicitation", "streamable-http"),
],
indirect=True,
)
async def test_elicitation(server_transport: str, server_url: str) -> None:
"""Test elicitation (user interaction) functionality.""" """Test elicitation (user interaction) functionality."""
transport = server_transport async with Client(elicitation.mcp, elicitation_callback=elicitation_callback) as client:
client_cm = create_client_for_transport(transport, server_url) # Test booking with unavailable date (triggers elicitation)
booking_result = await client.call_tool(
"book_table",
{
"date": "2024-12-25", # Unavailable date
"time": "19:00",
"party_size": 4,
},
)
assert len(booking_result.content) == 1
assert isinstance(booking_result.content[0], TextContent)
assert "[SUCCESS] Booked for 2024-12-26" in booking_result.content[0].text
async with client_cm as (read_stream, write_stream): # Test booking with available date (no elicitation)
async with ClientSession(read_stream, write_stream, elicitation_callback=elicitation_callback) as session: booking_result = await client.call_tool(
# Test initialization "book_table",
result = await session.initialize() {
assert isinstance(result, InitializeResult) "date": "2024-12-20", # Available date
assert result.server_info.name == "Elicitation Example" "time": "20:00",
"party_size": 2,
# Test booking with unavailable date (triggers elicitation) },
booking_result = await session.call_tool( )
"book_table", assert len(booking_result.content) == 1
{ assert isinstance(booking_result.content[0], TextContent)
"date": "2024-12-25", # Unavailable date assert "[SUCCESS] Booked for 2024-12-20 at 20:00" in booking_result.content[0].text
"time": "19:00",
"party_size": 4,
},
)
assert len(booking_result.content) == 1
assert isinstance(booking_result.content[0], TextContent)
assert "[SUCCESS] Booked for 2024-12-26" in booking_result.content[0].text
# Test booking with available date (no elicitation)
booking_result = await session.call_tool(
"book_table",
{
"date": "2024-12-20", # Available date
"time": "20:00",
"party_size": 2,
},
)
assert len(booking_result.content) == 1
assert isinstance(booking_result.content[0], TextContent)
assert "[SUCCESS] Booked for 2024-12-20 at 20:00" in booking_result.content[0].text
# Test notifications async def test_notifications() -> None:
@pytest.mark.anyio
@pytest.mark.parametrize(
"server_transport",
[
("notifications", "sse"),
("notifications", "streamable-http"),
],
indirect=True,
)
async def test_notifications(server_transport: str, server_url: str) -> None:
"""Test notifications and logging functionality.""" """Test notifications and logging functionality."""
transport = server_transport
collector = NotificationCollector() collector = NotificationCollector()
async def message_handler(message: RequestResponder[ServerRequest, ClientResult] | ServerNotification | Exception): async def message_handler(message: RequestResponder[ServerRequest, ClientResult] | ServerNotification | Exception):
@ -493,150 +272,87 @@ async def test_notifications(server_transport: str, server_url: str) -> None:
if isinstance(message, Exception): # pragma: no cover if isinstance(message, Exception): # pragma: no cover
raise message raise message
client_cm = create_client_for_transport(transport, server_url) async with Client(notifications.mcp, message_handler=message_handler) as client:
# Call tool that generates notifications
tool_result = await client.call_tool("process_data", {"data": "test_data"})
assert len(tool_result.content) == 1
assert isinstance(tool_result.content[0], TextContent)
assert "Processed: test_data" in tool_result.content[0].text
async with client_cm as (read_stream, write_stream): # Verify log messages at different levels
async with ClientSession(read_stream, write_stream, message_handler=message_handler) as session: assert len(collector.log_messages) >= 4
# Test initialization log_levels = {msg.level for msg in collector.log_messages}
result = await session.initialize() assert "debug" in log_levels
assert isinstance(result, InitializeResult) assert "info" in log_levels
assert result.server_info.name == "Notifications Example" assert "warning" in log_levels
assert "error" in log_levels
# Call tool that generates notifications # Verify resource list changed notification
tool_result = await session.call_tool("process_data", {"data": "test_data"}) assert len(collector.resource_notifications) > 0
assert len(tool_result.content) == 1
assert isinstance(tool_result.content[0], TextContent)
assert "Processed: test_data" in tool_result.content[0].text
# Verify log messages at different levels
assert len(collector.log_messages) >= 4
log_levels = {msg.level for msg in collector.log_messages}
assert "debug" in log_levels
assert "info" in log_levels
assert "warning" in log_levels
assert "error" in log_levels
# Verify resource list changed notification
assert len(collector.resource_notifications) > 0
# Test completion async def test_completion() -> None:
@pytest.mark.anyio
@pytest.mark.parametrize(
"server_transport",
[
("completion", "sse"),
("completion", "streamable-http"),
],
indirect=True,
)
async def test_completion(server_transport: str, server_url: str) -> None:
"""Test completion (autocomplete) functionality.""" """Test completion (autocomplete) functionality."""
transport = server_transport async with Client(completion.mcp) as client:
client_cm = create_client_for_transport(transport, server_url) assert client.server_capabilities is not None
assert client.server_capabilities.resources is not None
assert client.server_capabilities.prompts is not None
async with client_cm as (read_stream, write_stream): # Test resource completion
async with ClientSession(read_stream, write_stream) as session: completion_result = await client.complete(
# Test initialization ref=ResourceTemplateReference(type="ref/resource", uri="github://repos/{owner}/{repo}"),
result = await session.initialize() argument={"name": "repo", "value": ""},
assert isinstance(result, InitializeResult) context_arguments={"owner": "modelcontextprotocol"},
assert result.server_info.name == "Example" )
assert result.capabilities.resources is not None
assert result.capabilities.prompts is not None
# Test resource completion assert completion_result is not None
completion_result = await session.complete( assert hasattr(completion_result, "completion")
ref=ResourceTemplateReference(type="ref/resource", uri="github://repos/{owner}/{repo}"), assert completion_result.completion is not None
argument={"name": "repo", "value": ""}, assert len(completion_result.completion.values) == 3
context_arguments={"owner": "modelcontextprotocol"}, assert "python-sdk" in completion_result.completion.values
) assert "typescript-sdk" in completion_result.completion.values
assert "specification" in completion_result.completion.values
assert completion_result is not None # Test prompt completion
assert hasattr(completion_result, "completion") completion_result = await client.complete(
assert completion_result.completion is not None ref=PromptReference(type="ref/prompt", name="review_code"),
assert len(completion_result.completion.values) == 3 argument={"name": "language", "value": "py"},
assert "python-sdk" in completion_result.completion.values )
assert "typescript-sdk" in completion_result.completion.values
assert "specification" in completion_result.completion.values
# Test prompt completion assert completion_result is not None
completion_result = await session.complete( assert hasattr(completion_result, "completion")
ref=PromptReference(type="ref/prompt", name="review_code"), assert completion_result.completion is not None
argument={"name": "language", "value": "py"}, assert "python" in completion_result.completion.values
) assert all(lang.startswith("py") for lang in completion_result.completion.values)
assert completion_result is not None
assert hasattr(completion_result, "completion")
assert completion_result.completion is not None
assert "python" in completion_result.completion.values
assert all(lang.startswith("py") for lang in completion_result.completion.values)
# Test MCPServer quickstart example async def test_mcpserver_quickstart() -> None:
@pytest.mark.anyio
@pytest.mark.parametrize(
"server_transport",
[
("mcpserver_quickstart", "sse"),
("mcpserver_quickstart", "streamable-http"),
],
indirect=True,
)
async def test_mcpserver_quickstart(server_transport: str, server_url: str) -> None:
"""Test MCPServer quickstart example.""" """Test MCPServer quickstart example."""
transport = server_transport async with Client(mcpserver_quickstart.mcp) as client:
client_cm = create_client_for_transport(transport, server_url) # Test add tool
tool_result = await client.call_tool("add", {"a": 10, "b": 20})
assert len(tool_result.content) == 1
assert isinstance(tool_result.content[0], TextContent)
assert tool_result.content[0].text == "30"
async with client_cm as (read_stream, write_stream): # Test greeting resource directly
async with ClientSession(read_stream, write_stream) as session: resource_result = await client.read_resource("greeting://Alice")
# Test initialization assert len(resource_result.contents) == 1
result = await session.initialize() assert isinstance(resource_result.contents[0], TextResourceContents)
assert isinstance(result, InitializeResult) assert resource_result.contents[0].text == "Hello, Alice!"
assert result.server_info.name == "Demo"
# Test add tool
tool_result = await session.call_tool("add", {"a": 10, "b": 20})
assert len(tool_result.content) == 1
assert isinstance(tool_result.content[0], TextContent)
assert tool_result.content[0].text == "30"
# Test greeting resource directly
resource_result = await session.read_resource("greeting://Alice")
assert len(resource_result.contents) == 1
assert isinstance(resource_result.contents[0], TextResourceContents)
assert resource_result.contents[0].text == "Hello, Alice!"
# Test structured output example async def test_structured_output() -> None:
@pytest.mark.anyio
@pytest.mark.parametrize(
"server_transport",
[
("structured_output", "sse"),
("structured_output", "streamable-http"),
],
indirect=True,
)
async def test_structured_output(server_transport: str, server_url: str) -> None:
"""Test structured output functionality.""" """Test structured output functionality."""
transport = server_transport async with Client(structured_output.mcp) as client:
client_cm = create_client_for_transport(transport, server_url) # Test get_weather tool
weather_result = await client.call_tool("get_weather", {"city": "New York"})
assert len(weather_result.content) == 1
assert isinstance(weather_result.content[0], TextContent)
async with client_cm as (read_stream, write_stream): # Check that the result contains expected weather data
async with ClientSession(read_stream, write_stream) as session: result_text = weather_result.content[0].text
# Test initialization assert "22.5" in result_text # temperature
result = await session.initialize() assert "sunny" in result_text # condition
assert isinstance(result, InitializeResult) assert "45" in result_text # humidity
assert result.server_info.name == "Structured Output Example" assert "5.2" in result_text # wind_speed
# Test get_weather tool
weather_result = await session.call_tool("get_weather", {"city": "New York"})
assert len(weather_result.content) == 1
assert isinstance(weather_result.content[0], TextContent)
# Check that the result contains expected weather data
result_text = weather_result.content[0].text
assert "22.5" in result_text # temperature
assert "sunny" in result_text # condition
assert "45" in result_text # humidity
assert "5.2" in result_text # wind_speed