github-featured/modelcontextprotocol--python-sdk
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

refactor: move inline imports to module level (#1893)

Browse Source
This commit is contained in:
Max Isbey
2026-01-16 15:54:08 +01:00
committed by GitHub
parent c06362126b
commit 5d80f4efc8
24 changed files with 58 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mcp/client/__main__.py
+1 -2
View File
@@ -1,6 +1,7 @@
import argparse
import logging
import sys
import warnings
from functools import partial
from urllib.parse import urlparse
@@ -15,8 +16,6 @@ from mcp.shared.message import SessionMessage
from mcp.shared.session import RequestResponder
if not sys.warnoptions:
import warnings
warnings.simplefilter("ignore")
logging.basicConfig(level=logging.INFO)
src/mcp/client/auth/extensions/client_credentials.py
+1 -2
View File
@@ -9,6 +9,7 @@ Provides OAuth providers for machine-to-machine authentication flows:
"""
import time
import warnings
from collections.abc import Awaitable, Callable
from typing import Any, Literal
from uuid import uuid4
@@ -409,8 +410,6 @@ class RFC7523OAuthClientProvider(OAuthClientProvider):
timeout: float = 300.0,
jwt_parameters: JWTParameters | None = None,
) -> None:
import warnings
warnings.warn(
"RFC7523OAuthClientProvider is deprecated. Use ClientCredentialsOAuthProvider "
"or PrivateKeyJWTOAuthProvider instead.",
src/mcp/server/__main__.py
+1 -2
View File
@@ -1,6 +1,7 @@
import importlib.metadata
import logging
import sys
import warnings
import anyio
@@ -10,8 +11,6 @@ from mcp.server.stdio import stdio_server
from mcp.types import ServerCapabilities
if not sys.warnoptions:
import warnings
warnings.simplefilter("ignore")
logging.basicConfig(level=logging.INFO)
src/mcp/server/lowlevel/server.py
+2 -3
View File
@@ -74,6 +74,7 @@ import logging
import warnings
from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
from importlib.metadata import version as importlib_version
from typing import Any, Generic, TypeAlias, cast
import anyio
@@ -173,9 +174,7 @@ class Server(Generic[LifespanResultT, RequestT]):
def pkg_version(package: str) -> str:
try:
from importlib.metadata import version
return version(package)
return importlib_version(package)
except Exception: # pragma: no cover
pass
tests/client/auth/extensions/test_client_credentials.py
+1 -2
View File
@@ -1,4 +1,5 @@
import urllib.parse
import warnings
import jwt
import pytest
@@ -60,8 +61,6 @@ def rfc7523_oauth_provider(client_metadata: OAuthClientMetadata, mock_storage: M
"""Mock callback handler."""
return "test_auth_code", "test_state"
import warnings
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
return RFC7523OAuthClientProvider(
tests/client/test_auth.py
+3 -8
View File
@@ -5,7 +5,7 @@ Tests for refactored OAuth client authentication implementation.
import base64
import time
from unittest import mock
from urllib.parse import unquote
from urllib.parse import parse_qs, quote, unquote, urlparse
import httpx
import pytest
@@ -27,6 +27,8 @@ from mcp.client.auth.utils import (
is_valid_client_metadata_url,
should_use_client_metadata_url,
)
from mcp.server.auth.routes import build_metadata
from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
from mcp.shared.auth import (
OAuthClientInformationFull,
OAuthClientMetadata,
@@ -758,8 +760,6 @@ class TestProtectedResourceMetadata:
content = request.content.decode()
assert "resource=" in content
# Check URL-encoded resource parameter
from urllib.parse import quote
expected_resource = quote(oauth_provider.context.get_resource_url(), safe="")
assert f"resource={expected_resource}" in content
@@ -1226,8 +1226,6 @@ class TestAuthFlow:
"%3A", ":"
).replace("+", " ")
# Extract state from redirect URL
from urllib.parse import parse_qs, urlparse
parsed = urlparse(url)
params = parse_qs(parsed.query)
captured_state = params.get("state", [None])[0]
@@ -1336,9 +1334,6 @@ def test_build_metadata(
registration_endpoint: str,
revocation_endpoint: str,
):
from mcp.server.auth.routes import build_metadata
from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
metadata = build_metadata(
issuer_url=AnyHttpUrl(issuer_url),
service_documentation_url=AnyHttpUrl(service_documentation_url),
tests/client/test_http_unicode.py
+9 -13
View File
@@ -7,12 +7,20 @@ Verifies that Unicode text is correctly transmitted and received in both directi
import multiprocessing
import socket
from collections.abc import Generator
from collections.abc import AsyncGenerator, Generator
from contextlib import asynccontextmanager
from typing import Any
import pytest
from starlette.applications import Starlette
from starlette.routing import Mount
import mcp.types as types
from mcp.client.session import ClientSession
from mcp.client.streamable_http import streamable_http_client
from mcp.server import Server
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.types import TextContent, Tool
from tests.test_helpers import wait_for_server
# Test constants with various Unicode characters
@@ -37,19 +45,7 @@ UNICODE_TEST_STRINGS = {
def run_unicode_server(port: int) -> None: # pragma: no cover
"""Run the Unicode test server in a separate process."""
# Import inside the function since this runs in a separate process
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from typing import Any
import uvicorn
from starlette.applications import Starlette
from starlette.routing import Mount
import mcp.types as types
from mcp.server import Server
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.types import TextContent, Tool
# Need to recreate the server setup in this process
server = Server(name="unicode_test_server")
tests/client/test_list_roots_callback.py
+1 -2
View File
@@ -2,6 +2,7 @@ import pytest
from pydantic import FileUrl
from mcp.client.session import ClientSession
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.server import Context
from mcp.server.session import ServerSession
from mcp.shared.context import RequestContext
@@ -13,8 +14,6 @@ from mcp.types import ListRootsResult, Root, TextContent
@pytest.mark.anyio
async def test_list_roots_callback():
from mcp.server.fastmcp import FastMCP
server = FastMCP("test")
callback_return = ListRootsResult(
tests/client/test_logging_callback.py
+1 -2
View File
@@ -3,6 +3,7 @@ from typing import Any, Literal
import pytest
import mcp.types as types
from mcp.server.fastmcp import FastMCP
from mcp.shared.memory import (
create_connected_server_and_client_session as create_session,
)
@@ -23,8 +24,6 @@ class LoggingCollector:
@pytest.mark.anyio
async def test_logging_callback():
from mcp.server.fastmcp import FastMCP
server = FastMCP("test")
logging_collector = LoggingCollector()
tests/client/test_output_schema_validation.py
+2 -4
View File
@@ -1,8 +1,10 @@
import inspect
import logging
from contextlib import contextmanager
from typing import Any
from unittest.mock import patch
import jsonschema
import pytest
from mcp.server.lowlevel import Server
@@ -19,15 +21,11 @@ def bypass_server_output_validation():
This simulates a malicious or non-compliant server that doesn't validate
its outputs, allowing us to test client-side validation.
"""
import jsonschema
# Save the original validate function
original_validate = jsonschema.validate
# Create a mock that tracks which module is calling it
def selective_mock(instance: Any = None, schema: Any = None, *args: Any, **kwargs: Any) -> None:
import inspect
# Check the call stack to see where this is being called from
for frame_info in inspect.stack():
# If called from the server module, skip validation
tests/client/test_sampling_callback.py
+1 -4
View File
@@ -1,6 +1,7 @@
import pytest
from mcp.client.session import ClientSession
from mcp.server.fastmcp import FastMCP
from mcp.shared.context import RequestContext
from mcp.shared.memory import (
create_connected_server_and_client_session as create_session,
@@ -17,8 +18,6 @@ from mcp.types import (
@pytest.mark.anyio
async def test_sampling_callback():
from mcp.server.fastmcp import FastMCP
server = FastMCP("test")
callback_return = CreateMessageResult(
@@ -63,8 +62,6 @@ async def test_sampling_callback():
@pytest.mark.anyio
async def test_create_message_backwards_compat_single_content():
"""Test backwards compatibility: create_message without tools returns single content."""
from mcp.server.fastmcp import FastMCP
server = FastMCP("test")
# Callback returns single content (text)
tests/client/test_session_group.py
+1 -2
View File
@@ -1,6 +1,7 @@
import contextlib
from unittest import mock
import httpx
import pytest
import mcp
@@ -356,8 +357,6 @@ class TestClientSessionGroup:
assert isinstance(server_params_instance, StreamableHttpParameters)
# Verify streamable_http_client was called with url, httpx_client, and terminate_on_close
# The http_client is created by the real create_mcp_http_client
import httpx
call_args = mock_specific_client_func.call_args
assert call_args.kwargs["url"] == server_params_instance.url
assert call_args.kwargs["terminate_on_close"] == server_params_instance.terminate_on_close
tests/client/test_stdio.py
+6 -7
View File
@@ -10,7 +10,12 @@ import anyio
import pytest
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, _create_platform_compatible_process, stdio_client
from mcp.client.stdio import (
StdioServerParameters,
_create_platform_compatible_process,
_terminate_process_tree,
stdio_client,
)
from mcp.shared.exceptions import McpError
from mcp.shared.message import SessionMessage
from mcp.types import CONNECTION_CLOSED, JSONRPCMessage, JSONRPCRequest, JSONRPCResponse
@@ -312,8 +317,6 @@ class TestChildProcessCleanup:
# Terminate using our function
print("Terminating process and children...")
from mcp.client.stdio import _terminate_process_tree
await _terminate_process_tree(proc)
# Verify processes stopped
@@ -413,8 +416,6 @@ class TestChildProcessCleanup:
assert new_size > initial_size, f"{name} process should be writing"
# Terminate the whole tree
from mcp.client.stdio import _terminate_process_tree
await _terminate_process_tree(proc)
# Verify all stopped
@@ -494,8 +495,6 @@ class TestChildProcessCleanup:
assert size2 > size1, "Child should be writing"
# Terminate - this will kill the process group even if parent exits first
from mcp.client.stdio import _terminate_process_tree
await _terminate_process_tree(proc)
# Verify child stopped
tests/server/auth/test_error_handling.py
+4 -6
View File
@@ -2,6 +2,9 @@
Tests for OAuth error handling in the auth handlers.
"""
import base64
import hashlib
import secrets
import unittest.mock
from typing import Any
from urllib.parse import parse_qs, urlparse
@@ -14,6 +17,7 @@ from starlette.applications import Starlette
from mcp.server.auth.provider import AuthorizeError, RegistrationError, TokenError
from mcp.server.auth.routes import create_auth_routes
from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
from tests.server.fastmcp.auth.test_auth_integration import MockOAuthProvider
@@ -25,8 +29,6 @@ def oauth_provider():
@pytest.fixture
def app(oauth_provider: MockOAuthProvider):
from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
# Enable client registration
client_registration_options = ClientRegistrationOptions(enabled=True)
revocation_options = RevocationOptions(enabled=True)
@@ -53,10 +55,6 @@ def client(app: Starlette):
@pytest.fixture
def pkce_challenge():
"""Create a PKCE challenge with code_verifier and code_challenge."""
import base64
import hashlib
import secrets
# Generate a code verifier
code_verifier = secrets.token_urlsafe(64)[:128]
tests/server/auth/test_protected_resource.py
+2 -4
View File
@@ -2,6 +2,8 @@
Integration tests for MCP Oauth Protected Resource.
"""
from urllib.parse import urlparse
import httpx
import pytest
from inline_snapshot import snapshot
@@ -159,8 +161,6 @@ class TestRouteConsistency:
)
# Extract path from metadata URL
from urllib.parse import urlparse
metadata_path = urlparse(str(metadata_url)).path
# Verify consistency
@@ -181,8 +181,6 @@ class TestRouteConsistency:
# Test URL generation
metadata_url = build_resource_metadata_url(resource_url_obj)
from urllib.parse import urlparse
url_path = urlparse(str(metadata_url)).path
# Test route creation
tests/server/fastmcp/auth/test_auth_integration.py
-4
View File
@@ -1258,8 +1258,6 @@ class TestAuthEndpoints:
)
# Send base64 without colon (invalid format)
import base64
invalid_creds = base64.b64encode(b"no-colon-here").decode()
response = await test_client.post(
"/token",
@@ -1306,8 +1304,6 @@ class TestAuthEndpoints:
)
# Send different client_id in Basic auth header
import base64
wrong_creds = base64.b64encode(f"wrong-client-id:{client_info['client_secret']}".encode()).decode()
response = await test_client.post(
"/token",
tests/server/fastmcp/test_func_metadata.py
+2 -12
View File
@@ -5,13 +5,14 @@
# pyright: reportUnknownLambdaType=false
from collections.abc import Callable
from dataclasses import dataclass
from typing import Annotated, Any, Final, TypedDict
from typing import Annotated, Any, Final, NamedTuple, TypedDict
import annotated_types
import pytest
from dirty_equals import IsPartialDict
from pydantic import BaseModel, Field
from mcp.server.fastmcp.exceptions import InvalidSignature
from mcp.server.fastmcp.utilities.func_metadata import func_metadata
from mcp.types import CallToolResult
@@ -558,7 +559,6 @@ async def test_str_annotation_runtime_validation():
def test_structured_output_requires_return_annotation():
"""Test that structured_output=True requires a return annotation"""
from mcp.server.fastmcp.exceptions import InvalidSignature
def func_no_annotation(): # pragma: no cover
return "hello"
@@ -881,8 +881,6 @@ def test_tool_call_result_annotated_is_structured_and_invalid():
def test_tool_call_result_in_optional_is_rejected():
"""Test that Optional[CallToolResult] raises InvalidSignature"""
from mcp.server.fastmcp.exceptions import InvalidSignature
def func_optional_call_tool_result() -> CallToolResult | None: # pragma: no cover
return CallToolResult(content=[])
@@ -896,8 +894,6 @@ def test_tool_call_result_in_optional_is_rejected():
def test_tool_call_result_in_union_is_rejected():
"""Test that Union[str, CallToolResult] raises InvalidSignature"""
from mcp.server.fastmcp.exceptions import InvalidSignature
def func_union_call_tool_result() -> str | CallToolResult: # pragma: no cover
return CallToolResult(content=[])
@@ -910,7 +906,6 @@ def test_tool_call_result_in_union_is_rejected():
def test_tool_call_result_in_pipe_union_is_rejected():
"""Test that str | CallToolResult raises InvalidSignature"""
from mcp.server.fastmcp.exceptions import InvalidSignature
def func_pipe_union_call_tool_result() -> str | CallToolResult: # pragma: no cover
return CallToolResult(content=[])
@@ -985,9 +980,6 @@ def test_structured_output_nested_models():
def test_structured_output_unserializable_type_error():
"""Test error when structured_output=True is used with unserializable types"""
from typing import NamedTuple
from mcp.server.fastmcp.exceptions import InvalidSignature
# Test with a class that has non-serializable default values
class ConfigWithCallable:
@@ -1185,8 +1177,6 @@ def test_basemodel_reserved_names_with_json_preparsing():
def test_disallowed_type_qualifier():
from mcp.server.fastmcp.exceptions import InvalidSignature
def func_disallowed_qualifier() -> Final[int]: # type: ignore
pass # pragma: no cover
tests/server/fastmcp/test_integration.py
+2 -6
View File
@@ -51,8 +51,10 @@ from mcp.types import (
NotificationParams,
ProgressNotification,
ProgressNotificationParams,
PromptReference,
ReadResourceResult,
ResourceListChangedNotification,
ResourceTemplateReference,
ServerNotification,
ServerRequest,
TextContent,
@@ -583,8 +585,6 @@ async def test_completion(server_transport: str, server_url: str) -> None:
assert result.capabilities.prompts is not None
# Test resource completion
from mcp.types import ResourceTemplateReference
completion_result = await session.complete(
ref=ResourceTemplateReference(type="ref/resource", uri="github://repos/{owner}/{repo}"),
argument={"name": "repo", "value": ""},
@@ -600,8 +600,6 @@ async def test_completion(server_transport: str, server_url: str) -> None:
assert "specification" in completion_result.completion.values
# Test prompt completion
from mcp.types import PromptReference
completion_result = await session.complete(
ref=PromptReference(type="ref/prompt", name="review_code"),
argument={"name": "language", "value": "py"},
@@ -644,8 +642,6 @@ async def test_fastmcp_quickstart(server_transport: str, server_url: str) -> Non
assert tool_result.content[0].text == "30"
# Test greeting resource directly
from pydantic import AnyUrl
resource_result = await session.read_resource(AnyUrl("greeting://Alice"))
assert len(resource_result.contents) == 1
assert isinstance(resource_result.contents[0], TextResourceContents)
tests/server/fastmcp/test_server.py
+2 -4
View File
@@ -5,9 +5,11 @@ from unittest.mock import patch
import pytest
from pydantic import BaseModel
from starlette.applications import Starlette
from starlette.routing import Mount, Route
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.fastmcp.exceptions import ToolError
from mcp.server.fastmcp.prompts.base import Message, UserMessage
from mcp.server.fastmcp.resources import FileResource, FunctionResource
from mcp.server.fastmcp.utilities.types import Audio, Image
@@ -51,8 +53,6 @@ class TestServer:
@pytest.mark.anyio
async def test_sse_app_returns_starlette_app(self):
"""Test that sse_app returns a Starlette application with correct routes."""
from starlette.applications import Starlette
mcp = FastMCP("test", host="0.0.0.0") # Use 0.0.0.0 to avoid auto DNS protection
app = mcp.sse_app()
@@ -617,8 +617,6 @@ class TestServerTools:
@pytest.mark.anyio
async def test_remove_nonexistent_tool(self):
"""Test that removing a non-existent tool raises ToolError."""
from mcp.server.fastmcp.exceptions import ToolError
mcp = FastMCP()
with pytest.raises(ToolError, match="Unknown tool: nonexistent"):
tests/server/fastmcp/test_url_elicitation.py
+2 -9
View File
@@ -2,10 +2,11 @@
import anyio
import pytest
from pydantic import BaseModel, Field
from mcp import types
from mcp.client.session import ClientSession
from mcp.server.elicitation import CancelledElicitation, DeclinedElicitation
from mcp.server.elicitation import CancelledElicitation, DeclinedElicitation, elicit_url
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
from mcp.shared.context import RequestContext
@@ -110,8 +111,6 @@ async def test_url_elicitation_cancel():
@pytest.mark.anyio
async def test_url_elicitation_helper_function():
"""Test the elicit_url helper function."""
from mcp.server.elicitation import elicit_url
mcp = FastMCP(name="URLElicitationHelperServer")
@mcp.tool(description="Tool using elicit_url helper")
@@ -180,8 +179,6 @@ async def test_url_no_content_in_response():
@pytest.mark.anyio
async def test_form_mode_still_works():
"""Ensure form mode elicitation still works after SEP 1036."""
from pydantic import BaseModel, Field
mcp = FastMCP(name="FormModeBackwardCompatServer")
class NameSchema(BaseModel):
@@ -267,8 +264,6 @@ async def test_url_elicitation_required_error_code():
@pytest.mark.anyio
async def test_elicit_url_typed_results():
"""Test that elicit_url returns properly typed result objects."""
from mcp.server.elicitation import elicit_url
mcp = FastMCP(name="TypedResultsServer")
@mcp.tool(description="Test declined result")
@@ -329,8 +324,6 @@ async def test_elicit_url_typed_results():
@pytest.mark.anyio
async def test_deprecated_elicit_method():
"""Test the deprecated elicit() method for backward compatibility."""
from pydantic import BaseModel, Field
mcp = FastMCP(name="DeprecatedElicitServer")
class EmailSchema(BaseModel):
tests/shared/test_auth_utils.py
+2 -2
View File
@@ -1,5 +1,7 @@
"""Tests for OAuth 2.0 Resource Indicators utilities."""
from pydantic import HttpUrl
from mcp.shared.auth_utils import check_resource_allowed, resource_url_from_server_url
@@ -37,8 +39,6 @@ class TestResourceUrlFromServerUrl:
def test_handles_pydantic_urls(self):
"""Should handle Pydantic URL types."""
from pydantic import HttpUrl
url = HttpUrl("https://example.com/path")
assert resource_url_from_server_url(url) == "https://example.com/path"
tests/shared/test_streamable_http.py
+1 -4
View File
@@ -10,6 +10,7 @@ import json
import multiprocessing
import socket
import time
import traceback
from collections.abc import Generator
from typing import Any
from unittest.mock import MagicMock
@@ -462,8 +463,6 @@ def run_server(
try:
server.run()
except Exception:
import traceback
traceback.print_exc()
@@ -1100,8 +1099,6 @@ async def test_streamable_http_client_json_response(json_response_server: None,
@pytest.mark.anyio
async def test_streamable_http_client_get_stream(basic_server: None, basic_server_url: str):
"""Test GET stream functionality for server-initiated messages."""
import mcp.types as types
notifications_received: list[types.ServerNotification] = []
# Define message handler to capture notifications
tests/shared/test_ws.py
+1 -2
View File
@@ -3,6 +3,7 @@ import socket
import time
from collections.abc import AsyncGenerator, Generator
from typing import Any
from urllib.parse import urlparse
import anyio
import pytest
@@ -50,8 +51,6 @@ class ServerTest(Server): # pragma: no cover
@self.read_resource()
async def handle_read_resource(uri: str) -> str | bytes:
from urllib.parse import urlparse
parsed = urlparse(uri)
if parsed.scheme == "foobar":
return f"Read {parsed.netloc}"
tests/test_examples.py
+10 -16
View File
@@ -6,10 +6,16 @@
# pyright: reportUnknownMemberType=false
import sys
from pathlib import Path
import pytest
from pydantic import AnyUrl
from pytest_examples import CodeExample, EvalExample, find_examples
from examples.fastmcp.complex_inputs import mcp as complex_inputs_mcp
from examples.fastmcp.desktop import mcp as desktop_mcp
from examples.fastmcp.direct_call_tool_result_return import mcp as direct_call_tool_result_mcp
from examples.fastmcp.simple_echo import mcp as simple_echo_mcp
from mcp.shared.memory import create_connected_server_and_client_session as client_session
from mcp.types import TextContent, TextResourceContents
@@ -17,9 +23,7 @@ from mcp.types import TextContent, TextResourceContents
@pytest.mark.anyio
async def test_simple_echo():
"""Test the simple echo server"""
from examples.fastmcp.simple_echo import mcp
async with client_session(mcp._mcp_server) as client:
async with client_session(simple_echo_mcp._mcp_server) as client:
result = await client.call_tool("echo", {"text": "hello"})
assert len(result.content) == 1
content = result.content[0]
@@ -30,9 +34,7 @@ async def test_simple_echo():
@pytest.mark.anyio
async def test_complex_inputs():
"""Test the complex inputs server"""
from examples.fastmcp.complex_inputs import mcp
async with client_session(mcp._mcp_server) as client:
async with client_session(complex_inputs_mcp._mcp_server) as client:
tank = {"shrimp": [{"name": "bob"}, {"name": "alice"}]}
result = await client.call_tool("name_shrimp", {"tank": tank, "extra_names": ["charlie"]})
assert len(result.content) == 3
@@ -47,9 +49,7 @@ async def test_complex_inputs():
@pytest.mark.anyio
async def test_direct_call_tool_result_return():
"""Test the CallToolResult echo server"""
from examples.fastmcp.direct_call_tool_result_return import mcp
async with client_session(mcp._mcp_server) as client:
async with client_session(direct_call_tool_result_mcp._mcp_server) as client:
result = await client.call_tool("echo", {"text": "hello"})
assert len(result.content) == 1
content = result.content[0]
@@ -64,18 +64,12 @@ async def test_direct_call_tool_result_return():
@pytest.mark.anyio
async def test_desktop(monkeypatch: pytest.MonkeyPatch):
"""Test the desktop server"""
from pathlib import Path
from pydantic import AnyUrl
from examples.fastmcp.desktop import mcp
# Mock desktop directory listing
mock_files = [Path("/fake/path/file1.txt"), Path("/fake/path/file2.txt")]
monkeypatch.setattr(Path, "iterdir", lambda self: mock_files) # type: ignore[reportUnknownArgumentType]
monkeypatch.setattr(Path, "home", lambda: Path("/fake/home"))
async with client_session(mcp._mcp_server) as client:
async with client_session(desktop_mcp._mcp_server) as client:
# Test the sum function
result = await client.call_tool("sum", {"a": 1, "b": 2})
assert len(result.content) == 1