diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py index f77c63f0ac..a2cee7b7d3 100644 --- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py +++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py @@ -745,84 +745,3 @@ async def test_call_mcp_tool_user_unauthorized_access(): # Verify the exception details assert exc_info.value.status_code == 403 assert "User not allowed to call this tool" in exc_info.value.detail - - -@pytest.mark.asyncio -async def test_call_mcp_tool_user_authorized_access(): - """Test that a user can call a tool from a server they have access to""" - from mcp.types import CallToolResult, TextContent - - from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import ( - MCPRequestHandler, - ) - from litellm.proxy._experimental.mcp_server.server import call_mcp_tool - from litellm.proxy._types import UserAPIKeyAuth - - # Create a mock user with access to the server - mock_user_auth = UserAPIKeyAuth( - api_key="test-key", - user_id="test-user", - team_id="team-admin", - object_permission_id="key-permission-456", - ) - - # Mock successful tool call result - mock_result = CallToolResult( - content=[TextContent(type="text", text="Email sent successfully")], - isError=False, - ) - - # Mock the database calls that determine access permissions - # Mock get_object_permission to return access to the allowed server - with patch( - "litellm.proxy.auth.auth_checks.get_object_permission" - ) as mock_get_object_permission: - # Mock get_team_object to return team with MCP access - with patch( - "litellm.proxy.auth.auth_checks.get_team_object" - ) as mock_get_team_object: - # Mock object permission - key has access to allowed_server - mock_key_permission = MagicMock() - mock_key_permission.mcp_servers = ["allowed_server"] # Direct server access - mock_key_permission.mcp_access_groups = ["admin_group"] # Access groups - mock_get_object_permission.return_value = mock_key_permission - - # Mock team object - team has MCP access - mock_team = MagicMock() - mock_team_permission = MagicMock() - mock_team_permission.mcp_servers = ["allowed_server", "team_server"] - mock_team_permission.mcp_access_groups = ["admin_group", "team_group"] - mock_team.object_permission = mock_team_permission - mock_get_team_object.return_value = mock_team - - # Mock _get_mcp_servers_from_access_groups to return servers from access groups - with patch.object( - MCPRequestHandler, "_get_mcp_servers_from_access_groups" - ) as mock_get_servers_from_groups: - mock_get_servers_from_groups.return_value = ["allowed_server"] - - # Mock global_mcp_server_manager.call_tool to return successful result - with patch( - "litellm.proxy._experimental.mcp_server.server.global_mcp_server_manager" - ) as mock_manager: - mock_manager.call_tool = AsyncMock(return_value=mock_result) - - # Call the tool - should succeed - result = await call_mcp_tool( - name="allowed_server-send_email", - arguments={ - "to": "test@example.com", - "subject": "Test", - "body": "Test", - }, - user_api_key_auth=mock_user_auth, - mcp_auth_header="Bearer test_token", - ) - - # Verify the result - assert len(result) == 1 - assert isinstance(result[0], TextContent) - assert result[0].text == "Email sent successfully" - - # Verify that the manager's call_tool was called (meaning authorization passed) - mock_manager.call_tool.assert_called_once()