mirror of
https://github.com/mcp-use/mcp-use.git
synced 2026-09-14 18:06:01 +08:00
Luigi Pederzani
b45d9663ec
* fix(python): expose client capabilities to MCP servers Add list_roots_callback support to Python connectors to properly advertise the roots capability during MCP initialization handshake. Previously, the Python library was not exposing any client capabilities because the list_roots_callback was not being passed to ClientSession. This fix ensures the roots capability is always advertised, bringing the Python library in line with the TypeScript implementation. Changes: - Add roots and list_roots_callback params to BaseConnector - Add get_roots() and set_roots() methods for managing roots - Pass list_roots_callback to ClientSession in all connectors - Export Root type from connectors module - Update unit tests to expect list_roots_callback parameter Closes MCP-1063 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * feat(python): add roots and list_roots_callback params to connectors - Add roots and list_roots_callback parameters to StdioConnector, HttpConnector, and SandboxConnector constructors - Add comprehensive unit tests for client capabilities exposure - Tests verify roots capability, get_roots(), set_roots(), and that all callbacks are correctly passed to ClientSession 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * chore: fix import sorting and update API documentation - Fix import sorting in test_client_capabilities.py (ruff) - Regenerate API documentation for updated connector classes 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * feat(python): add list_roots() method to server Context Allow MCP servers to request the list of roots from connected clients via ctx.list_roots(). This follows the same pattern as ctx.sample() and ctx.elicit() for server-to-client requests. * feat(python): add roots params to MCPClient and config - Add roots and list_roots_callback parameters to MCPClient.__init__, from_dict(), and from_config_file() methods - Update create_connector_from_config() to pass roots params to all connector types (Stdio, HTTP, Sandbox) - Allows clients to configure roots at the MCPClient level and have them propagate to all sessions * test(python): add get_client_roots tool to primitive server Add a tool that demonstrates the roots capability by requesting and returning the list of roots from the connected client. Used for integration testing of the roots feature. * test(python): add integration tests for roots capability Add comprehensive integration tests for the roots feature: - test_roots_capability: Verify server can request/receive roots - test_roots_empty_by_default: Verify empty roots when not provided - test_roots_with_custom_callback: Verify custom callback precedence - test_set_roots_updates_server_view: Verify dynamic root updates * test(python): update unit tests for roots parameters Update mock assertions in test_client.py to expect the new roots and list_roots_callback parameters in create_connector_from_config calls. * ci(python): add roots to primitive tests and README badge - Add roots to CI workflow primitive test matrix - Add roots to PR comment report primitives list - Add Roots badge to README Primitives row * feat(docs): enhance MCP client and server documentation for roots feature - Added `roots` and `list_roots_callback` parameters to MCP client initialization documentation. - Updated `create_connector_from_config` documentation to include new parameters. - Introduced `list_roots` method documentation in server context, detailing its usage and return values. - Improved overall clarity and completeness of API reference for better developer guidance. * feat(python): add get_client_roots tool to context example - Introduced a new tool, `get_client_roots`, to the context example, demonstrating how to retrieve and display the list of roots exposed by the client. - Updated documentation to include the new tool and its functionality, enhancing the example's coverage of context features. * feat(docs): add roots documentation for client and server - Introduced new documentation for the `roots` feature, detailing how clients can expose file system roots to MCP servers and how servers can access these roots. - Added examples for configuring roots in the MCP client and using the `list_roots()` method in the server context. - Enhanced overall clarity and usability of the documentation to support developers in implementing the roots functionality. --------- Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com> Co-authored-by: pietrozullo <pietro.zullo@gmail.com>
297 lines
11 KiB
Python
297 lines
11 KiB
Python
"""
|
|
Unit tests for MCP client capabilities exposure.
|
|
|
|
These tests verify that the mcp-use library correctly exposes client capabilities
|
|
to MCP servers during the initialization handshake.
|
|
"""
|
|
|
|
from unittest.mock import ANY, AsyncMock, MagicMock, Mock, patch
|
|
|
|
import pytest
|
|
from mcp.types import ListRootsResult, Root
|
|
|
|
from mcp_use.client.connectors.base import BaseConnector
|
|
from mcp_use.client.connectors.http import HttpConnector
|
|
from mcp_use.client.connectors.stdio import StdioConnector
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_logger():
|
|
"""Mock the logger to prevent errors during tests."""
|
|
with patch("mcp_use.client.connectors.base.logger") as mock_base_logger:
|
|
with patch("mcp_use.client.connectors.stdio.logger"):
|
|
with patch("mcp_use.client.connectors.http.logger"):
|
|
yield mock_base_logger
|
|
|
|
|
|
class TestRootsCapability:
|
|
"""Tests for roots capability exposure."""
|
|
|
|
def test_list_roots_callback_always_provided(self):
|
|
"""Test that list_roots_callback property always returns a callback.
|
|
|
|
This ensures the roots capability is always advertised to servers.
|
|
"""
|
|
connector = StdioConnector()
|
|
|
|
# list_roots_callback should always return a callable
|
|
callback = connector.list_roots_callback
|
|
assert callable(callback)
|
|
|
|
def test_default_roots_empty(self):
|
|
"""Test that roots are empty by default."""
|
|
connector = StdioConnector()
|
|
|
|
roots = connector.get_roots()
|
|
assert roots == []
|
|
|
|
def test_init_with_roots(self):
|
|
"""Test initialization with roots parameter."""
|
|
initial_roots = [
|
|
Root(uri="file:///home/user/project", name="My Project"),
|
|
Root(uri="file:///home/user/data"),
|
|
]
|
|
|
|
connector = StdioConnector(roots=initial_roots)
|
|
|
|
roots = connector.get_roots()
|
|
assert len(roots) == 2
|
|
assert str(roots[0].uri) == "file:///home/user/project"
|
|
assert roots[0].name == "My Project"
|
|
assert str(roots[1].uri) == "file:///home/user/data"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_roots_updates_cache(self):
|
|
"""Test that set_roots() updates the internal roots cache."""
|
|
connector = StdioConnector()
|
|
|
|
new_roots = [
|
|
Root(uri="file:///new/path", name="New Root"),
|
|
]
|
|
|
|
await connector.set_roots(new_roots)
|
|
|
|
roots = connector.get_roots()
|
|
assert len(roots) == 1
|
|
assert str(roots[0].uri) == "file:///new/path"
|
|
assert roots[0].name == "New Root"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_roots_sends_notification_when_connected(self):
|
|
"""Test that set_roots() sends notification when connected."""
|
|
connector = StdioConnector()
|
|
connector._connected = True
|
|
|
|
# Mock the client session
|
|
mock_session = MagicMock()
|
|
mock_session.send_roots_list_changed = AsyncMock()
|
|
connector.client_session = mock_session
|
|
|
|
new_roots = [Root(uri="file:///test/path")]
|
|
await connector.set_roots(new_roots)
|
|
|
|
# Verify notification was sent
|
|
mock_session.send_roots_list_changed.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_roots_no_notification_when_disconnected(self):
|
|
"""Test that set_roots() doesn't send notification when disconnected."""
|
|
connector = StdioConnector()
|
|
connector._connected = False
|
|
|
|
# Mock the client session
|
|
mock_session = MagicMock()
|
|
mock_session.send_roots_list_changed = AsyncMock()
|
|
connector.client_session = mock_session
|
|
|
|
new_roots = [Root(uri="file:///test/path")]
|
|
await connector.set_roots(new_roots)
|
|
|
|
# Verify notification was NOT sent
|
|
mock_session.send_roots_list_changed.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_internal_list_roots_callback_returns_roots(self):
|
|
"""Test that the internal callback returns the cached roots."""
|
|
initial_roots = [
|
|
Root(uri="file:///home/user/project", name="My Project"),
|
|
]
|
|
|
|
connector = StdioConnector(roots=initial_roots)
|
|
|
|
# Create a mock context
|
|
mock_context = MagicMock()
|
|
|
|
# Call the internal callback
|
|
result = await connector._internal_list_roots_callback(mock_context)
|
|
|
|
assert isinstance(result, ListRootsResult)
|
|
assert len(result.roots) == 1
|
|
assert str(result.roots[0].uri) == "file:///home/user/project"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_custom_list_roots_callback_takes_precedence(self):
|
|
"""Test that a custom list_roots_callback takes precedence."""
|
|
custom_roots = [Root(uri="file:///custom/path", name="Custom")]
|
|
|
|
async def custom_callback(context):
|
|
return ListRootsResult(roots=custom_roots)
|
|
|
|
connector = StdioConnector(
|
|
roots=[Root(uri="file:///default/path")], # Default roots
|
|
list_roots_callback=custom_callback,
|
|
)
|
|
|
|
# Create a mock context
|
|
mock_context = MagicMock()
|
|
|
|
# Call the internal callback (which should delegate to custom)
|
|
result = await connector._internal_list_roots_callback(mock_context)
|
|
|
|
# Should return custom roots, not default
|
|
assert len(result.roots) == 1
|
|
assert str(result.roots[0].uri) == "file:///custom/path"
|
|
assert result.roots[0].name == "Custom"
|
|
|
|
def test_get_roots_returns_copy(self):
|
|
"""Test that get_roots() returns a copy, not the original list."""
|
|
initial_roots = [Root(uri="file:///test/path")]
|
|
connector = StdioConnector(roots=initial_roots)
|
|
|
|
roots1 = connector.get_roots()
|
|
roots2 = connector.get_roots()
|
|
|
|
# Should be equal but not the same object
|
|
assert roots1 == roots2
|
|
assert roots1 is not roots2
|
|
assert roots1 is not connector._roots
|
|
|
|
|
|
class TestCapabilitiesInClientSession:
|
|
"""Tests that verify capabilities are passed to ClientSession."""
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("mcp_use.client.connectors.stdio.StdioConnectionManager")
|
|
@patch("mcp_use.client.connectors.stdio.ClientSession")
|
|
async def test_stdio_passes_list_roots_callback(self, mock_client_session, mock_connection_manager):
|
|
"""Test that StdioConnector passes list_roots_callback to ClientSession."""
|
|
# Setup mocks
|
|
mock_manager_instance = Mock()
|
|
mock_manager_instance.start = AsyncMock(return_value=("read_stream", "write_stream"))
|
|
mock_connection_manager.return_value = mock_manager_instance
|
|
|
|
mock_client_instance = Mock()
|
|
mock_client_instance.__aenter__ = AsyncMock()
|
|
mock_client_session.return_value = mock_client_instance
|
|
|
|
# Create connector and connect
|
|
connector = StdioConnector()
|
|
await connector.connect()
|
|
|
|
# Verify list_roots_callback was passed
|
|
call_kwargs = mock_client_session.call_args
|
|
assert "list_roots_callback" in call_kwargs.kwargs or (
|
|
len(call_kwargs.args) > 4 and call_kwargs.args[4] is not None
|
|
)
|
|
|
|
# Get the actual callback that was passed
|
|
if call_kwargs.kwargs.get("list_roots_callback"):
|
|
passed_callback = call_kwargs.kwargs["list_roots_callback"]
|
|
else:
|
|
# Find it in positional args
|
|
passed_callback = None
|
|
for arg in call_kwargs.args:
|
|
if callable(arg) and arg.__name__ == "_internal_list_roots_callback":
|
|
passed_callback = arg
|
|
break
|
|
|
|
# The callback should be the connector's list_roots_callback
|
|
assert passed_callback is not None
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("mcp_use.client.connectors.http.StreamableHttpConnectionManager")
|
|
@patch("mcp_use.client.connectors.http.ClientSession")
|
|
async def test_http_passes_list_roots_callback(self, mock_client_session, mock_cm_class):
|
|
"""Test that HttpConnector passes list_roots_callback to ClientSession."""
|
|
# Setup mocks
|
|
mock_cm_instance = MagicMock()
|
|
mock_cm_instance.start = AsyncMock(return_value=("read_stream", "write_stream"))
|
|
mock_cm_instance.close = AsyncMock()
|
|
mock_cm_class.return_value = mock_cm_instance
|
|
|
|
mock_client_instance = MagicMock()
|
|
mock_client_instance.__aenter__ = AsyncMock()
|
|
mock_client_instance.__aexit__ = AsyncMock()
|
|
mock_client_instance.initialize = AsyncMock()
|
|
mock_init_result = MagicMock()
|
|
mock_init_result.capabilities = MagicMock(tools=False, resources=False, prompts=False)
|
|
mock_client_instance.initialize.return_value = mock_init_result
|
|
mock_client_session.return_value = mock_client_instance
|
|
|
|
# Create connector and connect
|
|
connector = HttpConnector("http://localhost:8000")
|
|
await connector.connect()
|
|
|
|
# Verify list_roots_callback was passed
|
|
mock_client_session.assert_called_once()
|
|
call_kwargs = mock_client_session.call_args
|
|
|
|
# Check kwargs
|
|
assert call_kwargs.kwargs.get("list_roots_callback") is not None
|
|
|
|
|
|
class TestSamplingAndElicitationCapabilities:
|
|
"""Tests for sampling and elicitation capability exposure."""
|
|
|
|
def test_sampling_callback_stored(self):
|
|
"""Test that sampling_callback is stored when provided."""
|
|
|
|
async def my_sampling_callback(params):
|
|
pass
|
|
|
|
connector = StdioConnector(sampling_callback=my_sampling_callback)
|
|
assert connector.sampling_callback is my_sampling_callback
|
|
|
|
def test_elicitation_callback_stored(self):
|
|
"""Test that elicitation_callback is stored when provided."""
|
|
|
|
async def my_elicitation_callback(params):
|
|
pass
|
|
|
|
connector = StdioConnector(elicitation_callback=my_elicitation_callback)
|
|
assert connector.elicitation_callback is my_elicitation_callback
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("mcp_use.client.connectors.stdio.StdioConnectionManager")
|
|
@patch("mcp_use.client.connectors.stdio.ClientSession")
|
|
async def test_all_callbacks_passed_to_client_session(self, mock_client_session, mock_connection_manager):
|
|
"""Test that all capability callbacks are passed to ClientSession."""
|
|
# Setup mocks
|
|
mock_manager_instance = Mock()
|
|
mock_manager_instance.start = AsyncMock(return_value=("read_stream", "write_stream"))
|
|
mock_connection_manager.return_value = mock_manager_instance
|
|
|
|
mock_client_instance = Mock()
|
|
mock_client_instance.__aenter__ = AsyncMock()
|
|
mock_client_session.return_value = mock_client_instance
|
|
|
|
# Create callbacks
|
|
async def sampling_cb(params):
|
|
pass
|
|
|
|
async def elicitation_cb(params):
|
|
pass
|
|
|
|
# Create connector with all callbacks
|
|
connector = StdioConnector(
|
|
sampling_callback=sampling_cb,
|
|
elicitation_callback=elicitation_cb,
|
|
)
|
|
await connector.connect()
|
|
|
|
# Verify all callbacks were passed
|
|
call_kwargs = mock_client_session.call_args.kwargs
|
|
assert call_kwargs.get("sampling_callback") is sampling_cb
|
|
assert call_kwargs.get("elicitation_callback") is elicitation_cb
|
|
assert call_kwargs.get("list_roots_callback") is not None
|